sched_ctx_hypervisor.c 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2011, 2012 INRIA
  4. *
  5. * StarPU is free software; you can redistribute it and/or modify
  6. * it under the terms of the GNU Lesser General Public License as published by
  7. * the Free Software Foundation; either version 2.1 of the License, or (at
  8. * your option) any later version.
  9. *
  10. * StarPU is distributed in the hope that it will be useful, but
  11. * WITHOUT ANY WARRANTY; without even the implied warranty of
  12. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  13. *
  14. * See the GNU Lesser General Public License in COPYING.LGPL for more details.
  15. */
  16. #include <sched_ctx_hypervisor_intern.h>
  17. unsigned imposed_resize = 0;
  18. struct starpu_performance_counters* perf_counters = NULL;
  19. static void notify_idle_cycle(unsigned sched_ctx, int worker, double idle_time);
  20. static void notify_pushed_task(unsigned sched_ctx, int worker);
  21. static void notify_poped_task(unsigned sched_ctx, int worker, double flops);
  22. static void notify_post_exec_hook(unsigned sched_ctx, int taskid);
  23. static void notify_idle_end(unsigned sched_ctx, int worker);
  24. static void notify_submitted_job(struct starpu_task *task, unsigned footprint);
  25. extern struct hypervisor_policy idle_policy;
  26. extern struct hypervisor_policy app_driven_policy;
  27. extern struct hypervisor_policy gflops_rate_policy;
  28. #ifdef HAVE_GLPK_H
  29. extern struct hypervisor_policy lp_policy;
  30. extern struct hypervisor_policy lp2_policy;
  31. extern struct hypervisor_policy lp3_policy;
  32. #endif
  33. static struct hypervisor_policy *predefined_policies[] = {
  34. &idle_policy,
  35. &app_driven_policy,
  36. #ifdef HAVE_GLPK_H
  37. &lp_policy,
  38. &lp2_policy,
  39. &lp3_policy,
  40. #endif
  41. &gflops_rate_policy
  42. };
  43. static void _load_hypervisor_policy(struct hypervisor_policy *policy)
  44. {
  45. STARPU_ASSERT(policy);
  46. #ifdef STARPU_VERBOSE
  47. if (policy->name)
  48. {
  49. _STARPU_DEBUG("Use %s hypervisor policy \n", policy->name);
  50. }
  51. #endif
  52. hypervisor.policy.name = policy->name;
  53. hypervisor.policy.handle_poped_task = policy->handle_poped_task;
  54. hypervisor.policy.handle_pushed_task = policy->handle_pushed_task;
  55. hypervisor.policy.handle_idle_cycle = policy->handle_idle_cycle;
  56. hypervisor.policy.handle_idle_end = policy->handle_idle_end;
  57. hypervisor.policy.handle_post_exec_hook = policy->handle_post_exec_hook;
  58. hypervisor.policy.handle_submitted_job = policy->handle_submitted_job;
  59. }
  60. static struct hypervisor_policy *_find_hypervisor_policy_from_name(const char *policy_name)
  61. {
  62. if (!policy_name)
  63. return NULL;
  64. unsigned i;
  65. for (i = 0; i < sizeof(predefined_policies)/sizeof(predefined_policies[0]); i++)
  66. {
  67. struct hypervisor_policy *p;
  68. p = predefined_policies[i];
  69. if (p->name)
  70. {
  71. if (strcmp(policy_name, p->name) == 0) {
  72. /* we found a policy with the requested name */
  73. return p;
  74. }
  75. }
  76. }
  77. fprintf(stderr, "Warning: hypervisor policy \"%s\" was not found, try \"help\" to get a list\n", policy_name);
  78. /* nothing was found */
  79. return NULL;
  80. }
  81. static struct hypervisor_policy *_select_hypervisor_policy(struct hypervisor_policy* hypervisor_policy)
  82. {
  83. struct hypervisor_policy *selected_policy = NULL;
  84. if(hypervisor_policy && hypervisor_policy->custom)
  85. return hypervisor_policy;
  86. /* we look if the application specified the name of a policy to load */
  87. const char *policy_name;
  88. if (hypervisor_policy && hypervisor_policy->name)
  89. {
  90. policy_name = hypervisor_policy->name;
  91. }
  92. else
  93. {
  94. policy_name = getenv("HYPERVISOR_POLICY");
  95. }
  96. if (policy_name)
  97. selected_policy = _find_hypervisor_policy_from_name(policy_name);
  98. /* Perhaps there was no policy that matched the name */
  99. if (selected_policy)
  100. return selected_policy;
  101. /* If no policy was specified, we use the idle policy as a default */
  102. return &idle_policy;
  103. }
  104. /* initializez the performance counters that starpu will use to retrive hints for resizing */
  105. struct starpu_performance_counters* sched_ctx_hypervisor_init(struct hypervisor_policy *hypervisor_policy)
  106. {
  107. hypervisor.min_tasks = 0;
  108. hypervisor.nsched_ctxs = 0;
  109. pthread_mutex_init(&act_hypervisor_mutex, NULL);
  110. int i;
  111. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  112. {
  113. hypervisor.resize[i] = 0;
  114. hypervisor.configurations[i] = NULL;
  115. hypervisor.sched_ctxs[i] = STARPU_NMAX_SCHED_CTXS;
  116. hypervisor.sched_ctx_w[i].sched_ctx = STARPU_NMAX_SCHED_CTXS;
  117. hypervisor.sched_ctx_w[i].config = NULL;
  118. hypervisor.sched_ctx_w[i].total_flops = 0.0;
  119. hypervisor.sched_ctx_w[i].submitted_flops = 0.0;
  120. hypervisor.sched_ctx_w[i].remaining_flops = 0.0;
  121. hypervisor.sched_ctx_w[i].start_time = 0.0;
  122. hypervisor.sched_ctx_w[i].resize_ack.receiver_sched_ctx = -1;
  123. hypervisor.sched_ctx_w[i].resize_ack.moved_workers = NULL;
  124. hypervisor.sched_ctx_w[i].resize_ack.nmoved_workers = 0;
  125. hypervisor.sched_ctx_w[i].resize_ack.acked_workers = NULL;
  126. pthread_mutex_init(&hypervisor.sched_ctx_w[i].mutex, NULL);
  127. int j;
  128. for(j = 0; j < STARPU_NMAXWORKERS; j++)
  129. {
  130. hypervisor.sched_ctx_w[i].current_idle_time[j] = 0.0;
  131. hypervisor.sched_ctx_w[i].pushed_tasks[j] = 0;
  132. hypervisor.sched_ctx_w[i].poped_tasks[j] = 0;
  133. hypervisor.sched_ctx_w[i].elapsed_flops[j] = 0.0;
  134. hypervisor.sched_ctx_w[i].total_elapsed_flops[j] = 0.0;
  135. }
  136. }
  137. struct hypervisor_policy *selected_hypervisor_policy = _select_hypervisor_policy(hypervisor_policy);
  138. _load_hypervisor_policy(selected_hypervisor_policy);
  139. perf_counters = (struct starpu_performance_counters*)malloc(sizeof(struct starpu_performance_counters));
  140. perf_counters->notify_idle_cycle = notify_idle_cycle;
  141. perf_counters->notify_pushed_task = notify_pushed_task;
  142. perf_counters->notify_poped_task = notify_poped_task;
  143. perf_counters->notify_post_exec_hook = notify_post_exec_hook;
  144. perf_counters->notify_idle_end = notify_idle_end;
  145. perf_counters->notify_submitted_job = notify_submitted_job;
  146. starpu_notify_hypervisor_exists();
  147. return perf_counters;
  148. }
  149. char* sched_ctx_hypervisor_get_policy()
  150. {
  151. return hypervisor.policy.name;
  152. }
  153. /* the user can forbid the resizing process*/
  154. void sched_ctx_hypervisor_stop_resize(unsigned sched_ctx)
  155. {
  156. imposed_resize = 1;
  157. hypervisor.resize[sched_ctx] = 0;
  158. }
  159. /* the user can restart the resizing process*/
  160. void sched_ctx_hypervisor_start_resize(unsigned sched_ctx)
  161. {
  162. imposed_resize = 1;
  163. hypervisor.resize[sched_ctx] = 1;
  164. }
  165. void sched_ctx_hypervisor_shutdown(void)
  166. {
  167. printf("shutdown\n");
  168. int i;
  169. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  170. {
  171. if(hypervisor.sched_ctxs[i] != STARPU_NMAX_SCHED_CTXS && hypervisor.nsched_ctxs > 0)
  172. {
  173. sched_ctx_hypervisor_stop_resize(hypervisor.sched_ctxs[i]);
  174. sched_ctx_hypervisor_unregister_ctx(hypervisor.sched_ctxs[i]);
  175. pthread_mutex_destroy(&hypervisor.sched_ctx_w[i].mutex);
  176. }
  177. }
  178. perf_counters->notify_idle_cycle = NULL;
  179. perf_counters->notify_pushed_task = NULL;
  180. perf_counters->notify_poped_task = NULL;
  181. perf_counters->notify_post_exec_hook = NULL;
  182. perf_counters->notify_idle_end = NULL;
  183. free(perf_counters);
  184. perf_counters = NULL;
  185. pthread_mutex_destroy(&act_hypervisor_mutex);
  186. }
  187. /* the hypervisor is in charge only of the contexts registered to it*/
  188. void sched_ctx_hypervisor_register_ctx(unsigned sched_ctx, double total_flops)
  189. {
  190. hypervisor.configurations[sched_ctx] = (struct starpu_htbl32_node*)malloc(sizeof(struct starpu_htbl32_node));
  191. hypervisor.resize_requests[sched_ctx] = (struct starpu_htbl32_node*)malloc(sizeof(struct starpu_htbl32_node));
  192. _add_config(sched_ctx);
  193. hypervisor.sched_ctx_w[sched_ctx].sched_ctx = sched_ctx;
  194. hypervisor.sched_ctxs[hypervisor.nsched_ctxs++] = sched_ctx;
  195. hypervisor.sched_ctx_w[sched_ctx].total_flops = total_flops;
  196. hypervisor.sched_ctx_w[sched_ctx].remaining_flops = total_flops;
  197. if(strcmp(hypervisor.policy.name, "app_driven") == 0)
  198. hypervisor.resize[sched_ctx] = 1;
  199. }
  200. static int _get_first_free_sched_ctx(int *sched_ctxs, unsigned nsched_ctxs)
  201. {
  202. int i;
  203. for(i = 0; i < nsched_ctxs; i++)
  204. if(sched_ctxs[i] == STARPU_NMAX_SCHED_CTXS)
  205. return i;
  206. return STARPU_NMAX_SCHED_CTXS;
  207. }
  208. /* rearange array of sched_ctxs in order not to have {MAXVAL, MAXVAL, 5, MAXVAL, 7}
  209. and have instead {5, 7, MAXVAL, MAXVAL, MAXVAL}
  210. it is easier afterwards to iterate the array
  211. */
  212. static void _rearange_sched_ctxs(int *sched_ctxs, int old_nsched_ctxs)
  213. {
  214. int first_free_id = STARPU_NMAX_SCHED_CTXS;
  215. int i;
  216. for(i = 0; i < old_nsched_ctxs; i++)
  217. {
  218. if(sched_ctxs[i] != STARPU_NMAX_SCHED_CTXS)
  219. {
  220. first_free_id = _get_first_free_sched_ctx(sched_ctxs, old_nsched_ctxs);
  221. if(first_free_id != STARPU_NMAX_SCHED_CTXS)
  222. {
  223. sched_ctxs[first_free_id] = sched_ctxs[i];
  224. sched_ctxs[i] = STARPU_NMAX_SCHED_CTXS;
  225. }
  226. }
  227. }
  228. }
  229. /* unregistered contexts will no longer be resized */
  230. void sched_ctx_hypervisor_unregister_ctx(unsigned sched_ctx)
  231. {
  232. unsigned i;
  233. for(i = 0; i < hypervisor.nsched_ctxs; i++)
  234. {
  235. if(hypervisor.sched_ctxs[i] == sched_ctx)
  236. {
  237. hypervisor.sched_ctxs[i] = STARPU_NMAX_SCHED_CTXS;
  238. break;
  239. }
  240. }
  241. _rearange_sched_ctxs(hypervisor.sched_ctxs, hypervisor.nsched_ctxs);
  242. hypervisor.nsched_ctxs--;
  243. hypervisor.sched_ctx_w[sched_ctx].sched_ctx = STARPU_NMAX_SCHED_CTXS;
  244. _remove_config(sched_ctx);
  245. free(hypervisor.configurations[sched_ctx]);
  246. free(hypervisor.resize_requests[sched_ctx]);
  247. }
  248. static int get_ntasks( int *tasks)
  249. {
  250. int ntasks = 0;
  251. int j;
  252. for(j = 0; j < STARPU_NMAXWORKERS; j++)
  253. {
  254. ntasks += tasks[j];
  255. }
  256. return ntasks;
  257. }
  258. static void _get_cpus(int *workers, int nworkers, int *cpus, int *ncpus)
  259. {
  260. int i, worker;
  261. *ncpus = 0;
  262. for(i = 0; i < nworkers; i++)
  263. {
  264. worker = workers[i];
  265. enum starpu_archtype arch = starpu_worker_get_type(worker);
  266. if(arch == STARPU_CPU_WORKER)
  267. cpus[(*ncpus)++] = worker;
  268. }
  269. }
  270. int get_nworkers_ctx(unsigned sched_ctx, enum starpu_archtype arch)
  271. {
  272. int nworkers_ctx = 0;
  273. struct worker_collection *workers = starpu_get_worker_collection_of_sched_ctx(sched_ctx);
  274. int worker;
  275. if(workers->init_cursor)
  276. workers->init_cursor(workers);
  277. while(workers->has_next(workers))
  278. {
  279. worker = workers->get_next(workers);
  280. enum starpu_archtype curr_arch = starpu_worker_get_type(worker);
  281. if(curr_arch == arch || arch == STARPU_ALL)
  282. nworkers_ctx++;
  283. }
  284. return nworkers_ctx;
  285. }
  286. /* actually move the workers: the cpus are moved, gpus are only shared */
  287. /* forbids another resize request before this one is take into account */
  288. void sched_ctx_hypervisor_move_workers(unsigned sender_sched_ctx, unsigned receiver_sched_ctx, int* workers_to_move, unsigned nworkers_to_move)
  289. {
  290. if(nworkers_to_move > 0 && hypervisor.resize[sender_sched_ctx])// && hypervisor.resize[receiver_sched_ctx])
  291. {
  292. int j;
  293. printf("resize ctx %d with", sender_sched_ctx);
  294. for(j = 0; j < nworkers_to_move; j++)
  295. printf(" %d", workers_to_move[j]);
  296. printf("\n");
  297. int *cpus = (int*) malloc(nworkers_to_move * sizeof(int));
  298. int ncpus;
  299. _get_cpus(workers_to_move, nworkers_to_move, cpus, &ncpus);
  300. // if(ncpus != 0)
  301. // starpu_remove_workers_from_sched_ctx(cpus, ncpus, sender_sched_ctx);
  302. starpu_remove_workers_from_sched_ctx(workers_to_move, nworkers_to_move, sender_sched_ctx);
  303. starpu_add_workers_to_sched_ctx(workers_to_move, nworkers_to_move, receiver_sched_ctx);
  304. pthread_mutex_lock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  305. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.receiver_sched_ctx = receiver_sched_ctx;
  306. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.moved_workers = (int*)malloc(nworkers_to_move * sizeof(int));
  307. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.nmoved_workers = nworkers_to_move;
  308. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.acked_workers = (int*)malloc(nworkers_to_move * sizeof(int));
  309. int i;
  310. for(i = 0; i < nworkers_to_move; i++)
  311. {
  312. hypervisor.sched_ctx_w[sender_sched_ctx].current_idle_time[workers_to_move[i]] = 0.0;
  313. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.moved_workers[i] = workers_to_move[i];
  314. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.acked_workers[i] = 0;
  315. }
  316. pthread_mutex_unlock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  317. hypervisor.resize[sender_sched_ctx] = 0;
  318. hypervisor.resize[receiver_sched_ctx] = 0;
  319. }
  320. return;
  321. }
  322. void sched_ctx_hypervisor_add_workers_to_sched_ctx(int* workers_to_add, unsigned nworkers_to_add, unsigned sched_ctx)
  323. {
  324. if(nworkers_to_add > 0 && hypervisor.resize[sched_ctx])
  325. {
  326. int j;
  327. printf("add to ctx %d:", sched_ctx);
  328. for(j = 0; j < nworkers_to_add; j++)
  329. printf(" %d", workers_to_add[j]);
  330. printf("\n");
  331. starpu_add_workers_to_sched_ctx(workers_to_add, nworkers_to_add, sched_ctx);
  332. pthread_mutex_lock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  333. hypervisor.sched_ctx_w[sched_ctx].resize_ack.receiver_sched_ctx = sched_ctx;
  334. hypervisor.sched_ctx_w[sched_ctx].resize_ack.moved_workers = (int*)malloc(nworkers_to_add * sizeof(int));
  335. hypervisor.sched_ctx_w[sched_ctx].resize_ack.nmoved_workers = nworkers_to_add;
  336. hypervisor.sched_ctx_w[sched_ctx].resize_ack.acked_workers = (int*)malloc(nworkers_to_add * sizeof(int));
  337. int i;
  338. for(i = 0; i < nworkers_to_add; i++)
  339. {
  340. hypervisor.sched_ctx_w[sched_ctx].current_idle_time[workers_to_add[i]] = 0.0;
  341. hypervisor.sched_ctx_w[sched_ctx].resize_ack.moved_workers[i] = workers_to_add[i];
  342. hypervisor.sched_ctx_w[sched_ctx].resize_ack.acked_workers[i] = 0;
  343. }
  344. pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  345. hypervisor.resize[sched_ctx] = 0;
  346. }
  347. return;
  348. }
  349. void sched_ctx_hypervisor_remove_workers_from_sched_ctx(int* workers_to_remove, unsigned nworkers_to_remove, unsigned sched_ctx)
  350. {
  351. if(nworkers_to_remove > 0 && hypervisor.resize[sched_ctx])
  352. {
  353. int j;
  354. printf("remove from ctx %d:", sched_ctx);
  355. for(j = 0; j < nworkers_to_remove; j++)
  356. printf(" %d", workers_to_remove[j]);
  357. printf("\n");
  358. starpu_remove_workers_from_sched_ctx(workers_to_remove, nworkers_to_remove, sched_ctx);
  359. /* hypervisor.sched_ctx_w[sched_ctx].resize_ack.receiver_sched_ctx = sched_ctx; */
  360. /* hypervisor.sched_ctx_w[sched_ctx].resize_ack.moved_workers = (int*)malloc(nworkers_to_remove * sizeof(int)); */
  361. /* hypervisor.sched_ctx_w[sched_ctx].resize_ack.nmoved_workers = nworkers_to_remove; */
  362. /* hypervisor.sched_ctx_w[sched_ctx].resize_ack.acked_workers = (int*)malloc(nworkers_to_remove * sizeof(int)); */
  363. /* int i; */
  364. /* for(i = 0; i < nworkers_to_remove; i++) */
  365. /* { */
  366. /* hypervisor.sched_ctx_w[sched_ctx].current_idle_time[workers_to_remove[i]] = 0.0; */
  367. /* hypervisor.sched_ctx_w[sched_ctx].resize_ack.moved_workers[i] = workers_to_remove[i]; */
  368. /* hypervisor.sched_ctx_w[sched_ctx].resize_ack.acked_workers[i] = 0; */
  369. /* } */
  370. // hypervisor.resize[sched_ctx] = 0;
  371. }
  372. return;
  373. }
  374. static void _set_elapsed_flops_per_sched_ctx(unsigned sched_ctx, double val)
  375. {
  376. int i;
  377. for(i = 0; i < STARPU_NMAXWORKERS; i++)
  378. hypervisor.sched_ctx_w[sched_ctx].elapsed_flops[i] = val;
  379. }
  380. double sched_ctx_hypervisor_get_elapsed_flops_per_sched_ctx(struct sched_ctx_wrapper* sc_w)
  381. {
  382. double ret_val = 0.0;
  383. int i;
  384. for(i = 0; i < STARPU_NMAXWORKERS; i++)
  385. ret_val += sc_w->elapsed_flops[i];
  386. return ret_val;
  387. }
  388. static unsigned _ack_resize_completed(unsigned sched_ctx, int worker)
  389. {
  390. struct resize_ack *resize_ack = NULL;
  391. unsigned sender_sched_ctx = STARPU_NMAX_SCHED_CTXS;
  392. if(hypervisor.nsched_ctxs > 0)
  393. {
  394. int i;
  395. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  396. {
  397. if(hypervisor.sched_ctxs[i] != STARPU_NMAX_SCHED_CTXS)
  398. {
  399. struct sched_ctx_wrapper *sc_w = &hypervisor.sched_ctx_w[hypervisor.sched_ctxs[i]];
  400. if(sc_w->resize_ack.receiver_sched_ctx != -1 &&
  401. sc_w->resize_ack.receiver_sched_ctx == sched_ctx)
  402. {
  403. resize_ack = &sc_w->resize_ack;
  404. sender_sched_ctx = hypervisor.sched_ctxs[i];
  405. break;
  406. }
  407. }
  408. }
  409. }
  410. /* if there is no ctx waiting for its ack return 1*/
  411. if(resize_ack == NULL)
  412. return 1;
  413. else
  414. {
  415. int *moved_workers = resize_ack->moved_workers;
  416. int nmoved_workers = resize_ack->nmoved_workers;
  417. int *acked_workers = resize_ack->acked_workers;
  418. int i;
  419. if(worker != -1)
  420. {
  421. for(i = 0; i < nmoved_workers; i++)
  422. {
  423. int moved_worker = moved_workers[i];
  424. if(moved_worker == worker && acked_workers[i] == 0)
  425. acked_workers[i] = 1;
  426. }
  427. }
  428. int nacked_workers = 0;
  429. for(i = 0; i < nmoved_workers; i++)
  430. {
  431. nacked_workers += (acked_workers[i] == 1);
  432. }
  433. unsigned resize_completed = (nacked_workers == nmoved_workers);
  434. unsigned receiver_sched_ctx = resize_ack->receiver_sched_ctx;
  435. /* if the permission to resize is not allowed by the user don't do it
  436. whatever the application says */
  437. if(resize_completed && !((hypervisor.resize[sched_ctx] == 0 || hypervisor.resize[receiver_sched_ctx] == 0) && imposed_resize) && worker == moved_workers[0])
  438. {
  439. /* info concerning only the gflops_rate strateg */
  440. struct sched_ctx_wrapper *sender_sc_w = &hypervisor.sched_ctx_w[sender_sched_ctx];
  441. struct sched_ctx_wrapper *receiver_sc_w = &hypervisor.sched_ctx_w[receiver_sched_ctx];
  442. double start_time = starpu_timing_now();
  443. sender_sc_w->start_time = start_time;
  444. sender_sc_w->remaining_flops = sender_sc_w->remaining_flops - sched_ctx_hypervisor_get_elapsed_flops_per_sched_ctx(sender_sc_w);
  445. _set_elapsed_flops_per_sched_ctx(sender_sched_ctx, 0.0);
  446. receiver_sc_w->start_time = start_time;
  447. receiver_sc_w->remaining_flops = receiver_sc_w->remaining_flops - sched_ctx_hypervisor_get_elapsed_flops_per_sched_ctx(receiver_sc_w);
  448. _set_elapsed_flops_per_sched_ctx(receiver_sched_ctx, 0.0);
  449. hypervisor.resize[sender_sched_ctx] = 1;
  450. hypervisor.resize[receiver_sched_ctx] = 1;
  451. /* if the user allowed resizing leave the decisions to the application */
  452. if(imposed_resize) imposed_resize = 0;
  453. pthread_mutex_lock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  454. resize_ack->receiver_sched_ctx = -1;
  455. resize_ack->nmoved_workers = 0;
  456. free(resize_ack->moved_workers);
  457. free(resize_ack->acked_workers);
  458. pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  459. }
  460. return resize_completed;
  461. }
  462. return 0;
  463. }
  464. void sched_ctx_hypervisor_resize(unsigned sched_ctx, int task_tag)
  465. {
  466. _starpu_htbl_insert_32(&hypervisor.resize_requests[sched_ctx], (uint32_t)task_tag, (void*)sched_ctx);
  467. }
  468. /* notifies the hypervisor that the worker is no longer idle and a new task was pushed on its queue */
  469. static void notify_idle_end(unsigned sched_ctx, int worker)
  470. {
  471. if(hypervisor.nsched_ctxs > 1)
  472. {
  473. if(hypervisor.resize[sched_ctx])
  474. hypervisor.sched_ctx_w[sched_ctx].current_idle_time[worker] = 0.0;
  475. if(hypervisor.policy.handle_idle_end)
  476. hypervisor.policy.handle_idle_end(sched_ctx, worker);
  477. _ack_resize_completed(sched_ctx, worker);
  478. }
  479. }
  480. /* notifies the hypervisor that the worker spent another cycle in idle time */
  481. static void notify_idle_cycle(unsigned sched_ctx, int worker, double idle_time)
  482. {
  483. if(hypervisor.nsched_ctxs > 1)
  484. {
  485. if(hypervisor.resize[sched_ctx])
  486. {
  487. struct sched_ctx_wrapper *sc_w = &hypervisor.sched_ctx_w[sched_ctx];
  488. sc_w->current_idle_time[worker] += idle_time;
  489. if(hypervisor.policy.handle_idle_cycle)
  490. hypervisor.policy.handle_idle_cycle(sched_ctx, worker);
  491. }
  492. else
  493. _ack_resize_completed(sched_ctx, worker);
  494. }
  495. return;
  496. }
  497. /* notifies the hypervisor that a new task was pushed on the queue of the worker */
  498. static void notify_pushed_task(unsigned sched_ctx, int worker)
  499. {
  500. hypervisor.sched_ctx_w[sched_ctx].pushed_tasks[worker]++;
  501. if(hypervisor.sched_ctx_w[sched_ctx].total_flops != 0.0 && hypervisor.sched_ctx_w[sched_ctx].start_time == 0.0)
  502. hypervisor.sched_ctx_w[sched_ctx].start_time = starpu_timing_now();
  503. int ntasks = get_ntasks(hypervisor.sched_ctx_w[sched_ctx].pushed_tasks);
  504. if(hypervisor.min_tasks == 0 || (!(hypervisor.resize[sched_ctx] == 0 && imposed_resize) && ntasks == hypervisor.min_tasks))
  505. {
  506. hypervisor.resize[sched_ctx] = 1;
  507. if(imposed_resize) imposed_resize = 0;
  508. }
  509. if(hypervisor.policy.handle_pushed_task)
  510. hypervisor.policy.handle_pushed_task(sched_ctx, worker);
  511. }
  512. /* notifies the hypervisor that a task was poped from the queue of the worker */
  513. static void notify_poped_task(unsigned sched_ctx, int worker, double elapsed_flops)
  514. {
  515. hypervisor.sched_ctx_w[sched_ctx].poped_tasks[worker]++;
  516. hypervisor.sched_ctx_w[sched_ctx].elapsed_flops[worker] += elapsed_flops;
  517. hypervisor.sched_ctx_w[sched_ctx].total_elapsed_flops[worker] += elapsed_flops;
  518. hypervisor.sched_ctx_w[sched_ctx].remaining_flops -= elapsed_flops; //sched_ctx_hypervisor_get_elapsed_flops_per_sched_ctx(&hypervisor.sched_ctx_w[sched_ctx]);
  519. if(hypervisor.nsched_ctxs > 1)
  520. {
  521. if(hypervisor.resize[sched_ctx])
  522. {
  523. if(hypervisor.policy.handle_poped_task)
  524. hypervisor.policy.handle_poped_task(sched_ctx, worker);
  525. }
  526. else
  527. _ack_resize_completed(sched_ctx, worker);
  528. }
  529. }
  530. /* notifies the hypervisor that a tagged task has just been executed */
  531. static void notify_post_exec_hook(unsigned sched_ctx, int task_tag)
  532. {
  533. STARPU_ASSERT(task_tag > 0);
  534. if(hypervisor.nsched_ctxs > 1)
  535. {
  536. unsigned conf_sched_ctx;
  537. int i;
  538. for(i = 0; i < hypervisor.nsched_ctxs; i++)
  539. {
  540. conf_sched_ctx = hypervisor.sched_ctxs[i];
  541. void *config = _starpu_htbl_search_32(hypervisor.configurations[conf_sched_ctx], (uint32_t)task_tag);
  542. if(config && config != hypervisor.configurations[conf_sched_ctx])
  543. {
  544. sched_ctx_hypervisor_set_config(conf_sched_ctx, config);
  545. free(config);
  546. _starpu_htbl_insert_32(&hypervisor.configurations[sched_ctx], (uint32_t)task_tag, NULL);
  547. }
  548. }
  549. /* for the app driven we have to wait for the resize to be available
  550. because the event is required to be executed at this specific moment */
  551. while(!_ack_resize_completed(sched_ctx, -1));
  552. if(hypervisor.resize[sched_ctx])
  553. {
  554. struct starpu_htbl32_node* resize_requests = hypervisor.resize_requests[sched_ctx];
  555. if(hypervisor.policy.handle_post_exec_hook)
  556. hypervisor.policy.handle_post_exec_hook(sched_ctx, resize_requests, task_tag);
  557. }
  558. }
  559. }
  560. static void notify_submitted_job(struct starpu_task *task, unsigned footprint)
  561. {
  562. pthread_mutex_lock(&act_hypervisor_mutex);
  563. hypervisor.sched_ctx_w[task->sched_ctx].submitted_flops += task->flops;
  564. pthread_mutex_unlock(&act_hypervisor_mutex);
  565. if(hypervisor.policy.handle_submitted_job)
  566. hypervisor.policy.handle_submitted_job(task, footprint);
  567. }
  568. struct sched_ctx_wrapper* sched_ctx_hypervisor_get_wrapper(unsigned sched_ctx)
  569. {
  570. return &hypervisor.sched_ctx_w[sched_ctx];
  571. }
  572. int* sched_ctx_hypervisor_get_sched_ctxs()
  573. {
  574. return hypervisor.sched_ctxs;
  575. }
  576. int sched_ctx_hypervisor_get_nsched_ctxs()
  577. {
  578. return hypervisor.nsched_ctxs;
  579. }