sched_ctx_hypervisor.c 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771
  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.size_ctxs = policy->size_ctxs;
  54. hypervisor.policy.handle_poped_task = policy->handle_poped_task;
  55. hypervisor.policy.handle_pushed_task = policy->handle_pushed_task;
  56. hypervisor.policy.handle_idle_cycle = policy->handle_idle_cycle;
  57. hypervisor.policy.handle_idle_end = policy->handle_idle_end;
  58. hypervisor.policy.handle_post_exec_hook = policy->handle_post_exec_hook;
  59. hypervisor.policy.handle_submitted_job = policy->handle_submitted_job;
  60. }
  61. static struct hypervisor_policy *_find_hypervisor_policy_from_name(const char *policy_name)
  62. {
  63. if (!policy_name)
  64. return NULL;
  65. unsigned i;
  66. for (i = 0; i < sizeof(predefined_policies)/sizeof(predefined_policies[0]); i++)
  67. {
  68. struct hypervisor_policy *p;
  69. p = predefined_policies[i];
  70. if (p->name)
  71. {
  72. if (strcmp(policy_name, p->name) == 0) {
  73. /* we found a policy with the requested name */
  74. return p;
  75. }
  76. }
  77. }
  78. fprintf(stderr, "Warning: hypervisor policy \"%s\" was not found, try \"help\" to get a list\n", policy_name);
  79. /* nothing was found */
  80. return NULL;
  81. }
  82. static struct hypervisor_policy *_select_hypervisor_policy(struct hypervisor_policy* hypervisor_policy)
  83. {
  84. struct hypervisor_policy *selected_policy = NULL;
  85. if(hypervisor_policy && hypervisor_policy->custom)
  86. return hypervisor_policy;
  87. /* we look if the application specified the name of a policy to load */
  88. const char *policy_name;
  89. if (hypervisor_policy && hypervisor_policy->name)
  90. {
  91. policy_name = hypervisor_policy->name;
  92. }
  93. else
  94. {
  95. policy_name = getenv("HYPERVISOR_POLICY");
  96. }
  97. if (policy_name)
  98. selected_policy = _find_hypervisor_policy_from_name(policy_name);
  99. /* Perhaps there was no policy that matched the name */
  100. if (selected_policy)
  101. return selected_policy;
  102. /* If no policy was specified, we use the idle policy as a default */
  103. return &idle_policy;
  104. }
  105. /* initializez the performance counters that starpu will use to retrive hints for resizing */
  106. struct starpu_performance_counters* sched_ctx_hypervisor_init(struct hypervisor_policy *hypervisor_policy)
  107. {
  108. hypervisor.min_tasks = 0;
  109. hypervisor.nsched_ctxs = 0;
  110. pthread_mutex_init(&act_hypervisor_mutex, NULL);
  111. int i;
  112. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  113. {
  114. hypervisor.resize[i] = 0;
  115. hypervisor.configurations[i] = NULL;
  116. hypervisor.sr = NULL;
  117. hypervisor.check_min_tasks[i] = 1;
  118. hypervisor.sched_ctxs[i] = STARPU_NMAX_SCHED_CTXS;
  119. hypervisor.sched_ctx_w[i].sched_ctx = STARPU_NMAX_SCHED_CTXS;
  120. hypervisor.sched_ctx_w[i].config = NULL;
  121. hypervisor.sched_ctx_w[i].total_flops = 0.0;
  122. hypervisor.sched_ctx_w[i].submitted_flops = 0.0;
  123. hypervisor.sched_ctx_w[i].remaining_flops = 0.0;
  124. hypervisor.sched_ctx_w[i].start_time = 0.0;
  125. hypervisor.sched_ctx_w[i].resize_ack.receiver_sched_ctx = -1;
  126. hypervisor.sched_ctx_w[i].resize_ack.moved_workers = NULL;
  127. hypervisor.sched_ctx_w[i].resize_ack.nmoved_workers = 0;
  128. hypervisor.sched_ctx_w[i].resize_ack.acked_workers = NULL;
  129. pthread_mutex_init(&hypervisor.sched_ctx_w[i].mutex, NULL);
  130. int j;
  131. for(j = 0; j < STARPU_NMAXWORKERS; j++)
  132. {
  133. hypervisor.sched_ctx_w[i].current_idle_time[j] = 0.0;
  134. hypervisor.sched_ctx_w[i].pushed_tasks[j] = 0;
  135. hypervisor.sched_ctx_w[i].poped_tasks[j] = 0;
  136. hypervisor.sched_ctx_w[i].elapsed_flops[j] = 0.0;
  137. hypervisor.sched_ctx_w[i].total_elapsed_flops[j] = 0.0;
  138. hypervisor.sched_ctx_w[i].worker_to_be_removed[j] = 0;
  139. }
  140. }
  141. struct hypervisor_policy *selected_hypervisor_policy = _select_hypervisor_policy(hypervisor_policy);
  142. _load_hypervisor_policy(selected_hypervisor_policy);
  143. perf_counters = (struct starpu_performance_counters*)malloc(sizeof(struct starpu_performance_counters));
  144. perf_counters->notify_idle_cycle = notify_idle_cycle;
  145. perf_counters->notify_pushed_task = notify_pushed_task;
  146. perf_counters->notify_poped_task = notify_poped_task;
  147. perf_counters->notify_post_exec_hook = notify_post_exec_hook;
  148. perf_counters->notify_idle_end = notify_idle_end;
  149. perf_counters->notify_submitted_job = notify_submitted_job;
  150. starpu_notify_hypervisor_exists();
  151. return perf_counters;
  152. }
  153. char* sched_ctx_hypervisor_get_policy()
  154. {
  155. return hypervisor.policy.name;
  156. }
  157. /* the user can forbid the resizing process*/
  158. void sched_ctx_hypervisor_stop_resize(unsigned sched_ctx)
  159. {
  160. imposed_resize = 1;
  161. hypervisor.resize[sched_ctx] = 0;
  162. }
  163. /* the user can restart the resizing process*/
  164. void sched_ctx_hypervisor_start_resize(unsigned sched_ctx)
  165. {
  166. imposed_resize = 1;
  167. hypervisor.resize[sched_ctx] = 1;
  168. }
  169. void sched_ctx_hypervisor_shutdown(void)
  170. {
  171. printf("shutdown\n");
  172. int i;
  173. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  174. {
  175. if(hypervisor.sched_ctxs[i] != STARPU_NMAX_SCHED_CTXS && hypervisor.nsched_ctxs > 0)
  176. {
  177. sched_ctx_hypervisor_stop_resize(hypervisor.sched_ctxs[i]);
  178. sched_ctx_hypervisor_unregister_ctx(hypervisor.sched_ctxs[i]);
  179. pthread_mutex_destroy(&hypervisor.sched_ctx_w[i].mutex);
  180. }
  181. }
  182. perf_counters->notify_idle_cycle = NULL;
  183. perf_counters->notify_pushed_task = NULL;
  184. perf_counters->notify_poped_task = NULL;
  185. perf_counters->notify_post_exec_hook = NULL;
  186. perf_counters->notify_idle_end = NULL;
  187. free(perf_counters);
  188. perf_counters = NULL;
  189. pthread_mutex_destroy(&act_hypervisor_mutex);
  190. }
  191. /* the hypervisor is in charge only of the contexts registered to it*/
  192. void sched_ctx_hypervisor_register_ctx(unsigned sched_ctx, double total_flops)
  193. {
  194. pthread_mutex_lock(&act_hypervisor_mutex);
  195. hypervisor.configurations[sched_ctx] = (struct starpu_htbl32_node*)malloc(sizeof(struct starpu_htbl32_node));
  196. hypervisor.resize_requests[sched_ctx] = (struct starpu_htbl32_node*)malloc(sizeof(struct starpu_htbl32_node));
  197. _add_config(sched_ctx);
  198. hypervisor.sched_ctx_w[sched_ctx].sched_ctx = sched_ctx;
  199. hypervisor.sched_ctxs[hypervisor.nsched_ctxs++] = sched_ctx;
  200. hypervisor.sched_ctx_w[sched_ctx].total_flops = total_flops;
  201. hypervisor.sched_ctx_w[sched_ctx].remaining_flops = total_flops;
  202. if(strcmp(hypervisor.policy.name, "app_driven") == 0)
  203. hypervisor.resize[sched_ctx] = 1;
  204. pthread_mutex_unlock(&act_hypervisor_mutex);
  205. }
  206. static int _get_first_free_sched_ctx(int *sched_ctxs, unsigned nsched_ctxs)
  207. {
  208. int i;
  209. for(i = 0; i < nsched_ctxs; i++)
  210. if(sched_ctxs[i] == STARPU_NMAX_SCHED_CTXS)
  211. return i;
  212. return STARPU_NMAX_SCHED_CTXS;
  213. }
  214. /* rearange array of sched_ctxs in order not to have {MAXVAL, MAXVAL, 5, MAXVAL, 7}
  215. and have instead {5, 7, MAXVAL, MAXVAL, MAXVAL}
  216. it is easier afterwards to iterate the array
  217. */
  218. static void _rearange_sched_ctxs(int *sched_ctxs, int old_nsched_ctxs)
  219. {
  220. int first_free_id = STARPU_NMAX_SCHED_CTXS;
  221. int i;
  222. for(i = 0; i < old_nsched_ctxs; i++)
  223. {
  224. if(sched_ctxs[i] != STARPU_NMAX_SCHED_CTXS)
  225. {
  226. first_free_id = _get_first_free_sched_ctx(sched_ctxs, old_nsched_ctxs);
  227. if(first_free_id != STARPU_NMAX_SCHED_CTXS)
  228. {
  229. sched_ctxs[first_free_id] = sched_ctxs[i];
  230. sched_ctxs[i] = STARPU_NMAX_SCHED_CTXS;
  231. }
  232. }
  233. }
  234. }
  235. /* unregistered contexts will no longer be resized */
  236. void sched_ctx_hypervisor_unregister_ctx(unsigned sched_ctx)
  237. {
  238. pthread_mutex_lock(&act_hypervisor_mutex);
  239. unsigned i;
  240. for(i = 0; i < hypervisor.nsched_ctxs; i++)
  241. {
  242. if(hypervisor.sched_ctxs[i] == sched_ctx)
  243. {
  244. hypervisor.sched_ctxs[i] = STARPU_NMAX_SCHED_CTXS;
  245. break;
  246. }
  247. }
  248. _rearange_sched_ctxs(hypervisor.sched_ctxs, hypervisor.nsched_ctxs);
  249. hypervisor.nsched_ctxs--;
  250. hypervisor.sched_ctx_w[sched_ctx].sched_ctx = STARPU_NMAX_SCHED_CTXS;
  251. _remove_config(sched_ctx);
  252. free(hypervisor.configurations[sched_ctx]);
  253. free(hypervisor.resize_requests[sched_ctx]);
  254. pthread_mutex_unlock(&act_hypervisor_mutex);
  255. }
  256. static int get_ntasks( int *tasks)
  257. {
  258. int ntasks = 0;
  259. int j;
  260. for(j = 0; j < STARPU_NMAXWORKERS; j++)
  261. {
  262. ntasks += tasks[j];
  263. }
  264. return ntasks;
  265. }
  266. static void _get_cpus(int *workers, int nworkers, int *cpus, int *ncpus)
  267. {
  268. int i, worker;
  269. *ncpus = 0;
  270. for(i = 0; i < nworkers; i++)
  271. {
  272. worker = workers[i];
  273. enum starpu_archtype arch = starpu_worker_get_type(worker);
  274. if(arch == STARPU_CPU_WORKER)
  275. cpus[(*ncpus)++] = worker;
  276. }
  277. }
  278. int get_nworkers_ctx(unsigned sched_ctx, enum starpu_archtype arch)
  279. {
  280. int nworkers_ctx = 0;
  281. struct worker_collection *workers = starpu_get_worker_collection_of_sched_ctx(sched_ctx);
  282. int worker;
  283. if(workers->init_cursor)
  284. workers->init_cursor(workers);
  285. while(workers->has_next(workers))
  286. {
  287. worker = workers->get_next(workers);
  288. enum starpu_archtype curr_arch = starpu_worker_get_type(worker);
  289. if(curr_arch == arch || arch == STARPU_ALL)
  290. nworkers_ctx++;
  291. }
  292. return nworkers_ctx;
  293. }
  294. /* actually move the workers: the cpus are moved, gpus are only shared */
  295. /* forbids another resize request before this one is take into account */
  296. void sched_ctx_hypervisor_move_workers(unsigned sender_sched_ctx, unsigned receiver_sched_ctx, int* workers_to_move, unsigned nworkers_to_move)
  297. {
  298. if(nworkers_to_move > 0 && hypervisor.resize[sender_sched_ctx])// && hypervisor.resize[receiver_sched_ctx])
  299. {
  300. int j;
  301. printf("resize ctx %d with", sender_sched_ctx);
  302. for(j = 0; j < nworkers_to_move; j++)
  303. printf(" %d", workers_to_move[j]);
  304. printf("\n");
  305. int *cpus = (int*) malloc(nworkers_to_move * sizeof(int));
  306. int ncpus;
  307. _get_cpus(workers_to_move, nworkers_to_move, cpus, &ncpus);
  308. // if(ncpus != 0)
  309. // starpu_remove_workers_from_sched_ctx(cpus, ncpus, sender_sched_ctx);
  310. starpu_add_workers_to_sched_ctx(workers_to_move, nworkers_to_move, receiver_sched_ctx);
  311. pthread_mutex_lock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  312. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.receiver_sched_ctx = receiver_sched_ctx;
  313. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.moved_workers = (int*)malloc(nworkers_to_move * sizeof(int));
  314. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.nmoved_workers = nworkers_to_move;
  315. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.acked_workers = (int*)malloc(nworkers_to_move * sizeof(int));
  316. int i;
  317. for(i = 0; i < nworkers_to_move; i++)
  318. {
  319. hypervisor.sched_ctx_w[sender_sched_ctx].current_idle_time[workers_to_move[i]] = 0.0;
  320. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.moved_workers[i] = workers_to_move[i];
  321. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.acked_workers[i] = 0;
  322. }
  323. hypervisor.resize[sender_sched_ctx] = 0;
  324. hypervisor.resize[receiver_sched_ctx] = 0;
  325. pthread_mutex_unlock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  326. }
  327. return;
  328. }
  329. void sched_ctx_hypervisor_add_workers_to_sched_ctx(int* workers_to_add, unsigned nworkers_to_add, unsigned sched_ctx)
  330. {
  331. if(nworkers_to_add > 0 && hypervisor.resize[sched_ctx])
  332. {
  333. int j;
  334. printf("add to ctx %d:", sched_ctx);
  335. for(j = 0; j < nworkers_to_add; j++)
  336. printf(" %d", workers_to_add[j]);
  337. printf("\n");
  338. int ret = pthread_mutex_trylock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  339. if(ret != EBUSY)
  340. {
  341. starpu_add_workers_to_sched_ctx(workers_to_add, nworkers_to_add, sched_ctx);
  342. pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  343. }
  344. }
  345. return;
  346. }
  347. unsigned sched_ctx_hypervisor_can_resize(unsigned sched_ctx)
  348. {
  349. return hypervisor.resize[sched_ctx];
  350. }
  351. void sched_ctx_hypervisor_remove_workers_from_sched_ctx(int* workers_to_remove, unsigned nworkers_to_remove, unsigned sched_ctx)
  352. {
  353. if(nworkers_to_remove > 0 && hypervisor.resize[sched_ctx])
  354. {
  355. int nworkers=0;
  356. int workers[nworkers_to_remove];
  357. int ret = pthread_mutex_trylock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  358. if(ret != EBUSY)
  359. {
  360. int i;
  361. for(i = 0; i < nworkers_to_remove; i++)
  362. if(starpu_worker_belongs_to_sched_ctx(workers_to_remove[i], sched_ctx))
  363. workers[nworkers++] = workers_to_remove[i];
  364. hypervisor.sched_ctx_w[sched_ctx].resize_ack.receiver_sched_ctx = -1;
  365. hypervisor.sched_ctx_w[sched_ctx].resize_ack.moved_workers = (int*)malloc(nworkers_to_remove * sizeof(int));
  366. hypervisor.sched_ctx_w[sched_ctx].resize_ack.nmoved_workers = nworkers;
  367. hypervisor.sched_ctx_w[sched_ctx].resize_ack.acked_workers = (int*)malloc(nworkers_to_remove * sizeof(int));
  368. for(i = 0; i < nworkers; i++)
  369. {
  370. hypervisor.sched_ctx_w[sched_ctx].current_idle_time[workers[i]] = 0.0;
  371. hypervisor.sched_ctx_w[sched_ctx].resize_ack.moved_workers[i] = workers[i];
  372. hypervisor.sched_ctx_w[sched_ctx].resize_ack.acked_workers[i] = 0;
  373. }
  374. hypervisor.resize[sched_ctx] = 0;
  375. pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  376. }
  377. }
  378. return;
  379. }
  380. static void _set_elapsed_flops_per_sched_ctx(unsigned sched_ctx, double val)
  381. {
  382. int i;
  383. for(i = 0; i < STARPU_NMAXWORKERS; i++)
  384. hypervisor.sched_ctx_w[sched_ctx].elapsed_flops[i] = val;
  385. }
  386. double sched_ctx_hypervisor_get_elapsed_flops_per_sched_ctx(struct sched_ctx_wrapper* sc_w)
  387. {
  388. double ret_val = 0.0;
  389. int i;
  390. for(i = 0; i < STARPU_NMAXWORKERS; i++)
  391. ret_val += sc_w->elapsed_flops[i];
  392. return ret_val;
  393. }
  394. double sched_ctx_hypervisor_get_total_elapsed_flops_per_sched_ctx(struct sched_ctx_wrapper* sc_w)
  395. {
  396. double ret_val = 0.0;
  397. int i;
  398. for(i = 0; i < STARPU_NMAXWORKERS; i++)
  399. ret_val += sc_w->total_elapsed_flops[i];
  400. return ret_val;
  401. }
  402. static unsigned _ack_resize_completed(unsigned sched_ctx, int worker)
  403. {
  404. if(!starpu_worker_belongs_to_sched_ctx(worker, sched_ctx))
  405. return 0;
  406. struct resize_ack *resize_ack = NULL;
  407. unsigned sender_sched_ctx = STARPU_NMAX_SCHED_CTXS;
  408. if(hypervisor.nsched_ctxs > 0)
  409. {
  410. int i;
  411. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  412. {
  413. if(hypervisor.sched_ctxs[i] != STARPU_NMAX_SCHED_CTXS)
  414. {
  415. struct sched_ctx_wrapper *sc_w = &hypervisor.sched_ctx_w[hypervisor.sched_ctxs[i]];
  416. unsigned only_remove = 0;
  417. if(sc_w->resize_ack.receiver_sched_ctx == -1 && hypervisor.sched_ctxs[i] != sched_ctx &&
  418. sc_w->resize_ack.nmoved_workers > 0 && starpu_worker_belongs_to_sched_ctx(worker, hypervisor.sched_ctxs[i]))
  419. {
  420. int j;
  421. for(j = 0; j < sc_w->resize_ack.nmoved_workers; j++)
  422. if(sc_w->resize_ack.moved_workers[j] == worker)
  423. {
  424. only_remove = 1;
  425. break;
  426. }
  427. }
  428. if(only_remove ||
  429. (sc_w->resize_ack.receiver_sched_ctx != -1 && sc_w->resize_ack.receiver_sched_ctx == sched_ctx))
  430. {
  431. resize_ack = &sc_w->resize_ack;
  432. sender_sched_ctx = hypervisor.sched_ctxs[i];
  433. break;
  434. }
  435. }
  436. }
  437. }
  438. /* if there is no ctx waiting for its ack return 1*/
  439. if(resize_ack == NULL)
  440. return 1;
  441. int *moved_workers = resize_ack->moved_workers;
  442. int nmoved_workers = resize_ack->nmoved_workers;
  443. int *acked_workers = resize_ack->acked_workers;
  444. int i;
  445. if(worker != -1)
  446. {
  447. int ret = pthread_mutex_trylock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  448. if(ret != EBUSY)
  449. {
  450. for(i = 0; i < nmoved_workers; i++)
  451. {
  452. int moved_worker = moved_workers[i];
  453. if(moved_worker == worker && acked_workers[i] == 0)
  454. {
  455. acked_workers[i] = 1;
  456. }
  457. }
  458. int nacked_workers = 0;
  459. for(i = 0; i < nmoved_workers; i++)
  460. {
  461. nacked_workers += (acked_workers[i] == 1);
  462. }
  463. unsigned resize_completed = (nacked_workers == nmoved_workers);
  464. int receiver_sched_ctx = sched_ctx;
  465. if(resize_completed)
  466. {
  467. /* if the permission to resize is not allowed by the user don't do it
  468. whatever the application says */
  469. if(!((hypervisor.resize[sender_sched_ctx] == 0 || hypervisor.resize[receiver_sched_ctx] == 0) && imposed_resize))
  470. {
  471. int j;
  472. printf("remove from ctx %d:", sender_sched_ctx);
  473. for(j = 0; j < nmoved_workers; j++)
  474. printf(" %d", moved_workers[j]);
  475. printf("\n");
  476. starpu_remove_workers_from_sched_ctx(moved_workers, nmoved_workers, sender_sched_ctx);
  477. /* info concerning only the gflops_rate strateg */
  478. struct sched_ctx_wrapper *sender_sc_w = &hypervisor.sched_ctx_w[sender_sched_ctx];
  479. struct sched_ctx_wrapper *receiver_sc_w = &hypervisor.sched_ctx_w[receiver_sched_ctx];
  480. double start_time = starpu_timing_now();
  481. sender_sc_w->start_time = start_time;
  482. sender_sc_w->remaining_flops = sender_sc_w->remaining_flops - sched_ctx_hypervisor_get_elapsed_flops_per_sched_ctx(sender_sc_w);
  483. _set_elapsed_flops_per_sched_ctx(sender_sched_ctx, 0.0);
  484. receiver_sc_w->start_time = start_time;
  485. receiver_sc_w->remaining_flops = receiver_sc_w->remaining_flops - sched_ctx_hypervisor_get_elapsed_flops_per_sched_ctx(receiver_sc_w);
  486. _set_elapsed_flops_per_sched_ctx(receiver_sched_ctx, 0.0);
  487. hypervisor.resize[sender_sched_ctx] = 1;
  488. hypervisor.resize[receiver_sched_ctx] = 1;
  489. /* if the user allowed resizing leave the decisions to the application */
  490. if(imposed_resize) imposed_resize = 0;
  491. resize_ack->receiver_sched_ctx = -1;
  492. resize_ack->nmoved_workers = 0;
  493. free(resize_ack->moved_workers);
  494. free(resize_ack->acked_workers);
  495. }
  496. pthread_mutex_unlock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  497. return resize_completed;
  498. }
  499. pthread_mutex_unlock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  500. return 0;
  501. }
  502. }
  503. return 0;
  504. }
  505. void sched_ctx_hypervisor_resize(unsigned sched_ctx, int task_tag)
  506. {
  507. _starpu_htbl_insert_32(&hypervisor.resize_requests[sched_ctx], (uint32_t)task_tag, (void*)sched_ctx);
  508. }
  509. /* notifies the hypervisor that the worker is no longer idle and a new task was pushed on its queue */
  510. static void notify_idle_end(unsigned sched_ctx, int worker)
  511. {
  512. if(hypervisor.nsched_ctxs > 1)
  513. {
  514. if(hypervisor.resize[sched_ctx])
  515. hypervisor.sched_ctx_w[sched_ctx].current_idle_time[worker] = 0.0;
  516. if(hypervisor.policy.handle_idle_end)
  517. hypervisor.policy.handle_idle_end(sched_ctx, worker);
  518. // _ack_resize_completed(sched_ctx, worker);
  519. }
  520. }
  521. /* notifies the hypervisor that the worker spent another cycle in idle time */
  522. static void notify_idle_cycle(unsigned sched_ctx, int worker, double idle_time)
  523. {
  524. if(hypervisor.nsched_ctxs > 1)
  525. {
  526. if(hypervisor.resize[sched_ctx])
  527. {
  528. struct sched_ctx_wrapper *sc_w = &hypervisor.sched_ctx_w[sched_ctx];
  529. sc_w->current_idle_time[worker] += idle_time;
  530. if(hypervisor.policy.handle_idle_cycle)
  531. hypervisor.policy.handle_idle_cycle(sched_ctx, worker);
  532. }
  533. /* else */
  534. /* _ack_resize_completed(sched_ctx, worker); */
  535. }
  536. return;
  537. }
  538. /* notifies the hypervisor that a new task was pushed on the queue of the worker */
  539. static void notify_pushed_task(unsigned sched_ctx, int worker)
  540. {
  541. hypervisor.sched_ctx_w[sched_ctx].pushed_tasks[worker]++;
  542. if(hypervisor.sched_ctx_w[sched_ctx].total_flops != 0.0 && hypervisor.sched_ctx_w[sched_ctx].start_time == 0.0)
  543. hypervisor.sched_ctx_w[sched_ctx].start_time = starpu_timing_now();
  544. int ntasks = get_ntasks(hypervisor.sched_ctx_w[sched_ctx].pushed_tasks);
  545. if((hypervisor.min_tasks == 0 || (!(hypervisor.resize[sched_ctx] == 0 && imposed_resize) && ntasks == hypervisor.min_tasks)) && hypervisor.check_min_tasks[sched_ctx])
  546. {
  547. hypervisor.resize[sched_ctx] = 1;
  548. if(imposed_resize) imposed_resize = 0;
  549. hypervisor.check_min_tasks[sched_ctx] = 0;
  550. }
  551. if(hypervisor.policy.handle_pushed_task)
  552. hypervisor.policy.handle_pushed_task(sched_ctx, worker);
  553. }
  554. /* notifies the hypervisor that a task was poped from the queue of the worker */
  555. static void notify_poped_task(unsigned sched_ctx, int worker, double elapsed_flops)
  556. {
  557. hypervisor.sched_ctx_w[sched_ctx].poped_tasks[worker]++;
  558. hypervisor.sched_ctx_w[sched_ctx].elapsed_flops[worker] += elapsed_flops;
  559. hypervisor.sched_ctx_w[sched_ctx].total_elapsed_flops[worker] += elapsed_flops;
  560. 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]);
  561. if(hypervisor.nsched_ctxs > 1)
  562. {
  563. if(hypervisor.resize[sched_ctx])
  564. {
  565. if(hypervisor.policy.handle_poped_task)
  566. hypervisor.policy.handle_poped_task(sched_ctx, worker);
  567. }
  568. _ack_resize_completed(sched_ctx, worker);
  569. }
  570. }
  571. /* notifies the hypervisor that a tagged task has just been executed */
  572. static void notify_post_exec_hook(unsigned sched_ctx, int task_tag)
  573. {
  574. STARPU_ASSERT(task_tag > 0);
  575. if(hypervisor.nsched_ctxs > 1)
  576. {
  577. unsigned conf_sched_ctx;
  578. int i;
  579. for(i = 0; i < hypervisor.nsched_ctxs; i++)
  580. {
  581. conf_sched_ctx = hypervisor.sched_ctxs[i];
  582. void *config = _starpu_htbl_search_32(hypervisor.configurations[conf_sched_ctx], (uint32_t)task_tag);
  583. if(config && config != hypervisor.configurations[conf_sched_ctx])
  584. {
  585. sched_ctx_hypervisor_set_config(conf_sched_ctx, config);
  586. free(config);
  587. _starpu_htbl_insert_32(&hypervisor.configurations[sched_ctx], (uint32_t)task_tag, NULL);
  588. }
  589. }
  590. /* for the app driven we have to wait for the resize to be available
  591. because the event is required to be executed at this specific moment */
  592. while(!_ack_resize_completed(sched_ctx, -1));
  593. if(hypervisor.resize[sched_ctx])
  594. {
  595. struct starpu_htbl32_node* resize_requests = hypervisor.resize_requests[sched_ctx];
  596. if(hypervisor.policy.handle_post_exec_hook)
  597. hypervisor.policy.handle_post_exec_hook(sched_ctx, resize_requests, task_tag);
  598. }
  599. }
  600. }
  601. static void notify_submitted_job(struct starpu_task *task, uint32_t footprint)
  602. {
  603. pthread_mutex_lock(&act_hypervisor_mutex);
  604. hypervisor.sched_ctx_w[task->sched_ctx].submitted_flops += task->flops;
  605. pthread_mutex_unlock(&act_hypervisor_mutex);
  606. if(hypervisor.policy.handle_submitted_job)
  607. hypervisor.policy.handle_submitted_job(task, footprint);
  608. }
  609. void sched_ctx_hypervisor_size_ctxs(int *sched_ctxs, int nsched_ctxs, int *workers, int nworkers)
  610. {
  611. int curr_nsched_ctxs = sched_ctxs == NULL ? hypervisor.nsched_ctxs : nsched_ctxs;
  612. int *curr_sched_ctxs = sched_ctxs == NULL ? hypervisor.sched_ctxs : sched_ctxs;
  613. int s;
  614. for(s = 0; s < curr_nsched_ctxs; s++)
  615. hypervisor.resize[curr_sched_ctxs[s]] = 1;
  616. if(hypervisor.policy.size_ctxs)
  617. hypervisor.policy.size_ctxs(curr_sched_ctxs, curr_nsched_ctxs, workers, nworkers);
  618. }
  619. struct sched_ctx_wrapper* sched_ctx_hypervisor_get_wrapper(unsigned sched_ctx)
  620. {
  621. return &hypervisor.sched_ctx_w[sched_ctx];
  622. }
  623. int* sched_ctx_hypervisor_get_sched_ctxs()
  624. {
  625. return hypervisor.sched_ctxs;
  626. }
  627. int sched_ctx_hypervisor_get_nsched_ctxs()
  628. {
  629. return hypervisor.nsched_ctxs;
  630. }
  631. void sched_ctx_hypervisor_save_size_req(int *sched_ctxs, int nsched_ctxs, int *workers, int nworkers)
  632. {
  633. hypervisor.sr = (struct size_request*)malloc(sizeof(struct size_request));
  634. hypervisor.sr->sched_ctxs = sched_ctxs;
  635. hypervisor.sr->nsched_ctxs = nsched_ctxs;
  636. hypervisor.sr->workers = workers;
  637. hypervisor.sr->nworkers = nworkers;
  638. }
  639. unsigned sched_ctx_hypervisor_get_size_req(int **sched_ctxs, int* nsched_ctxs, int **workers, int *nworkers)
  640. {
  641. if(hypervisor.sr != NULL)
  642. {
  643. *sched_ctxs = hypervisor.sr->sched_ctxs;
  644. *nsched_ctxs = hypervisor.sr->nsched_ctxs;
  645. *workers = hypervisor.sr->workers;
  646. *nworkers = hypervisor.sr->nworkers;
  647. return 1;
  648. }
  649. return 0;
  650. }
  651. void sched_ctx_hypervisor_free_size_req(void)
  652. {
  653. if(hypervisor.sr != NULL)
  654. {
  655. free(hypervisor.sr);
  656. hypervisor.sr = NULL;
  657. }
  658. }