sched_ctx_hypervisor.c 26 KB

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