sc_hypervisor.c 47 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410
  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 <sc_hypervisor_intern.h>
  17. #include <sc_hypervisor_policy.h>
  18. #include <starpu_config.h>
  19. unsigned imposed_resize = 0;
  20. unsigned type_of_tasks_known = 0;
  21. struct starpu_sched_ctx_performance_counters* perf_counters = NULL;
  22. static void notify_idle_cycle(unsigned sched_ctx, int worker, double idle_time);
  23. static void notify_pushed_task(unsigned sched_ctx, int worker);
  24. static void notify_post_exec_task(struct starpu_task *task, size_t data_size, uint32_t footprint,
  25. int hypervisor_tag, double flops);
  26. static void notify_poped_task(unsigned sched_ctx, int worker);
  27. static void notify_submitted_job(struct starpu_task *task, unsigned footprint, size_t data_size);
  28. static void notify_empty_ctx(unsigned sched_ctx, struct starpu_task *task);
  29. static void notify_delete_context(unsigned sched_ctx);
  30. extern struct sc_hypervisor_policy idle_policy;
  31. extern struct sc_hypervisor_policy app_driven_policy;
  32. extern struct sc_hypervisor_policy gflops_rate_policy;
  33. #ifdef STARPU_HAVE_GLPK_H
  34. extern struct sc_hypervisor_policy feft_lp_policy;
  35. extern struct sc_hypervisor_policy teft_lp_policy;
  36. extern struct sc_hypervisor_policy ispeed_lp_policy;
  37. extern struct sc_hypervisor_policy throughput_lp_policy;
  38. #endif // STARPU_HAVE_GLPK_
  39. extern struct sc_hypervisor_policy ispeed_policy;
  40. static struct sc_hypervisor_policy *predefined_policies[] =
  41. {
  42. &idle_policy,
  43. &app_driven_policy,
  44. #ifdef STARPU_HAVE_GLPK_H
  45. &feft_lp_policy,
  46. &teft_lp_policy,
  47. &ispeed_lp_policy,
  48. &throughput_lp_policy,
  49. #endif // STARPU_HAVE_GLPK_H
  50. &gflops_rate_policy,
  51. &ispeed_policy
  52. };
  53. static void _load_hypervisor_policy(struct sc_hypervisor_policy *policy)
  54. {
  55. STARPU_ASSERT(policy);
  56. hypervisor.policy.name = policy->name;
  57. hypervisor.policy.size_ctxs = policy->size_ctxs;
  58. hypervisor.policy.resize_ctxs = policy->resize_ctxs;
  59. hypervisor.policy.handle_poped_task = policy->handle_poped_task;
  60. hypervisor.policy.handle_pushed_task = policy->handle_pushed_task;
  61. hypervisor.policy.handle_idle_cycle = policy->handle_idle_cycle;
  62. hypervisor.policy.handle_idle_end = policy->handle_idle_end;
  63. hypervisor.policy.handle_post_exec_hook = policy->handle_post_exec_hook;
  64. hypervisor.policy.handle_submitted_job = policy->handle_submitted_job;
  65. hypervisor.policy.end_ctx = policy->end_ctx;
  66. }
  67. static struct sc_hypervisor_policy *_find_hypervisor_policy_from_name(const char *policy_name)
  68. {
  69. if (!policy_name)
  70. return NULL;
  71. unsigned i;
  72. for (i = 0; i < sizeof(predefined_policies)/sizeof(predefined_policies[0]); i++)
  73. {
  74. struct sc_hypervisor_policy *p;
  75. p = predefined_policies[i];
  76. if (p->name)
  77. {
  78. if (strcmp(policy_name, p->name) == 0) {
  79. /* we found a policy with the requested name */
  80. return p;
  81. }
  82. }
  83. }
  84. fprintf(stderr, "Warning: hypervisor policy \"%s\" was not found, try \"help\" to get a list\n", policy_name);
  85. /* nothing was found */
  86. return NULL;
  87. }
  88. static void display_sched_help_message(void)
  89. {
  90. const char* policy_name = getenv("SC_HYPERVISOR_POLICY");
  91. if (policy_name && (strcmp(policy_name, "help") == 0))
  92. {
  93. fprintf(stderr, "SC_HYPERVISOR_POLICY can be either of\n");
  94. /* display the description of all predefined policies */
  95. unsigned i;
  96. for (i = 0; i < sizeof(predefined_policies)/sizeof(predefined_policies[0]); i++)
  97. {
  98. struct sc_hypervisor_policy *p = predefined_policies[i];
  99. if (p->name)
  100. {
  101. fprintf(stderr, "%s\n", p->name);
  102. }
  103. }
  104. }
  105. }
  106. static struct sc_hypervisor_policy *_select_hypervisor_policy(struct sc_hypervisor_policy* hypervisor_policy)
  107. {
  108. struct sc_hypervisor_policy *selected_policy = NULL;
  109. if(hypervisor_policy && hypervisor_policy->custom)
  110. return hypervisor_policy;
  111. /* we look if the application specified the name of a policy to load */
  112. const char *policy_name;
  113. if (hypervisor_policy && hypervisor_policy->name)
  114. {
  115. policy_name = hypervisor_policy->name;
  116. }
  117. else
  118. {
  119. policy_name = getenv("SC_HYPERVISOR_POLICY");
  120. }
  121. if (policy_name)
  122. selected_policy = _find_hypervisor_policy_from_name(policy_name);
  123. /* Perhaps there was no policy that matched the name */
  124. if (selected_policy)
  125. return selected_policy;
  126. /* If no policy was specified, we use the idle policy as a default */
  127. return &idle_policy;
  128. }
  129. /* initializez the performance counters that starpu will use to retrive hints for resizing */
  130. void* sc_hypervisor_init(struct sc_hypervisor_policy *hypervisor_policy)
  131. {
  132. /* Perhaps we have to display some help */
  133. display_sched_help_message();
  134. hypervisor.min_tasks = 0;
  135. hypervisor.nsched_ctxs = 0;
  136. char* vel_gap = getenv("SC_HYPERVISOR_MAX_SPEED_GAP");
  137. hypervisor.max_speed_gap = vel_gap ? atof(vel_gap) : SC_SPEED_MAX_GAP_DEFAULT;
  138. char* crit = getenv("SC_HYPERVISOR_TRIGGER_RESIZE");
  139. hypervisor.resize_criteria = !crit ? SC_IDLE : strcmp(crit,"idle") == 0 ? SC_IDLE : (strcmp(crit,"speed") == 0 ? SC_SPEED : SC_NOTHING);
  140. starpu_pthread_mutex_init(&act_hypervisor_mutex, NULL);
  141. hypervisor.start_executing_time = starpu_timing_now();
  142. int i;
  143. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  144. {
  145. hypervisor.resize[i] = 0;
  146. hypervisor.allow_remove[i] = 1;
  147. hypervisor.configurations[i] = NULL;
  148. hypervisor.sr = NULL;
  149. hypervisor.check_min_tasks[i] = 1;
  150. hypervisor.sched_ctxs[i] = STARPU_NMAX_SCHED_CTXS;
  151. hypervisor.sched_ctx_w[i].sched_ctx = STARPU_NMAX_SCHED_CTXS;
  152. hypervisor.sched_ctx_w[i].config = NULL;
  153. hypervisor.sched_ctx_w[i].total_flops = 0.0;
  154. hypervisor.sched_ctx_w[i].submitted_flops = 0.0;
  155. hypervisor.sched_ctx_w[i].remaining_flops = 0.0;
  156. hypervisor.sched_ctx_w[i].start_time = 0.0;
  157. hypervisor.sched_ctx_w[i].real_start_time = 0.0;
  158. hypervisor.sched_ctx_w[i].hyp_react_start_time = 0.0;
  159. hypervisor.sched_ctx_w[i].resize_ack.receiver_sched_ctx = -1;
  160. hypervisor.sched_ctx_w[i].resize_ack.moved_workers = NULL;
  161. hypervisor.sched_ctx_w[i].resize_ack.nmoved_workers = 0;
  162. hypervisor.sched_ctx_w[i].resize_ack.acked_workers = NULL;
  163. starpu_pthread_mutex_init(&hypervisor.sched_ctx_w[i].mutex, NULL);
  164. hypervisor.optimal_v[i] = 0.0;
  165. hypervisor.sched_ctx_w[i].ref_speed[0] = -1.0;
  166. hypervisor.sched_ctx_w[i].ref_speed[1] = -1.0;
  167. hypervisor.sched_ctx_w[i].total_flops_available = 0;
  168. hypervisor.sched_ctx_w[i].to_be_sized = 0;
  169. int j;
  170. for(j = 0; j < STARPU_NMAXWORKERS; j++)
  171. {
  172. hypervisor.sched_ctx_w[i].start_time_w[i] = 0.0;
  173. hypervisor.sched_ctx_w[i].current_idle_time[j] = 0.0;
  174. hypervisor.sched_ctx_w[i].idle_time[j] = 0.0;
  175. hypervisor.sched_ctx_w[i].idle_start_time[j] = 0.0;
  176. hypervisor.sched_ctx_w[i].exec_time[j] = 0.0;
  177. hypervisor.sched_ctx_w[i].exec_start_time[j] = 0.0;
  178. hypervisor.sched_ctx_w[i].pushed_tasks[j] = 0;
  179. hypervisor.sched_ctx_w[i].poped_tasks[j] = 0;
  180. hypervisor.sched_ctx_w[i].elapsed_flops[j] = 0.0;
  181. hypervisor.sched_ctx_w[i].elapsed_data[j] = 0;
  182. hypervisor.sched_ctx_w[i].elapsed_tasks[j] = 0;
  183. hypervisor.sched_ctx_w[i].total_elapsed_flops[j] = 0.0;
  184. hypervisor.sched_ctx_w[i].worker_to_be_removed[j] = 0;
  185. hypervisor.sched_ctx_w[i].compute_idle[j] = 1;
  186. hypervisor.sched_ctx_w[i].compute_partial_idle[j] = 0;
  187. }
  188. }
  189. struct sc_hypervisor_policy *selected_hypervisor_policy = _select_hypervisor_policy(hypervisor_policy);
  190. _load_hypervisor_policy(selected_hypervisor_policy);
  191. perf_counters = (struct starpu_sched_ctx_performance_counters*)malloc(sizeof(struct starpu_sched_ctx_performance_counters));
  192. perf_counters->notify_idle_cycle = notify_idle_cycle;
  193. perf_counters->notify_pushed_task = notify_pushed_task;
  194. perf_counters->notify_poped_task = notify_poped_task;
  195. perf_counters->notify_post_exec_task = notify_post_exec_task;
  196. perf_counters->notify_submitted_job = notify_submitted_job;
  197. perf_counters->notify_empty_ctx = notify_empty_ctx;
  198. perf_counters->notify_delete_context = notify_delete_context;
  199. starpu_sched_ctx_notify_hypervisor_exists();
  200. return (void*)perf_counters;
  201. }
  202. const char* sc_hypervisor_get_policy()
  203. {
  204. return hypervisor.policy.name;
  205. }
  206. /* the user can forbid the resizing process*/
  207. void sc_hypervisor_stop_resize(unsigned sched_ctx)
  208. {
  209. imposed_resize = 1;
  210. hypervisor.resize[sched_ctx] = 0;
  211. }
  212. /* the user can restart the resizing process*/
  213. void sc_hypervisor_start_resize(unsigned sched_ctx)
  214. {
  215. imposed_resize = 1;
  216. hypervisor.resize[sched_ctx] = 1;
  217. }
  218. static void _print_current_time()
  219. {
  220. if(!getenv("SC_HYPERVISOR_STOP_PRINT"))
  221. {
  222. double curr_time = starpu_timing_now();
  223. double elapsed_time = (curr_time - hypervisor.start_executing_time) / 1000000.0; /* in seconds */
  224. fprintf(stdout, "Time: %lf\n", elapsed_time);
  225. int i;
  226. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  227. {
  228. if(hypervisor.sched_ctxs[i] != STARPU_NMAX_SCHED_CTXS)
  229. {
  230. struct sc_hypervisor_wrapper *sc_w = &hypervisor.sched_ctx_w[hypervisor.sched_ctxs[i]];
  231. double cpu_speed = sc_hypervisor_get_speed(sc_w, STARPU_CPU_WORKER);
  232. double cuda_speed = sc_hypervisor_get_speed(sc_w, STARPU_CUDA_WORKER);
  233. int ncpus = sc_hypervisor_get_nworkers_ctx(sc_w->sched_ctx, STARPU_CPU_WORKER);
  234. int ncuda = sc_hypervisor_get_nworkers_ctx(sc_w->sched_ctx, STARPU_CUDA_WORKER);
  235. fprintf(stdout, "%d: cpu_v = %lf cuda_v = %lf ncpus = %d ncuda = %d\n", hypervisor.sched_ctxs[i], cpu_speed, cuda_speed, ncpus, ncuda);
  236. }
  237. }
  238. }
  239. return;
  240. }
  241. void sc_hypervisor_shutdown(void)
  242. {
  243. int i;
  244. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  245. {
  246. if(hypervisor.sched_ctxs[i] != STARPU_NMAX_SCHED_CTXS && hypervisor.nsched_ctxs > 0)
  247. {
  248. sc_hypervisor_stop_resize(hypervisor.sched_ctxs[i]);
  249. sc_hypervisor_unregister_ctx(hypervisor.sched_ctxs[i]);
  250. starpu_pthread_mutex_destroy(&hypervisor.sched_ctx_w[i].mutex);
  251. }
  252. }
  253. perf_counters->notify_idle_cycle = NULL;
  254. perf_counters->notify_pushed_task = NULL;
  255. perf_counters->notify_poped_task = NULL;
  256. perf_counters->notify_post_exec_task = NULL;
  257. perf_counters->notify_delete_context = NULL;
  258. free(perf_counters);
  259. perf_counters = NULL;
  260. starpu_pthread_mutex_destroy(&act_hypervisor_mutex);
  261. }
  262. /* the hypervisor is in charge only of the contexts registered to it*/
  263. void sc_hypervisor_register_ctx(unsigned sched_ctx, double total_flops)
  264. {
  265. starpu_pthread_mutex_lock(&act_hypervisor_mutex);
  266. hypervisor.configurations[sched_ctx] = NULL;
  267. hypervisor.resize_requests[sched_ctx] = NULL;
  268. starpu_pthread_mutex_init(&hypervisor.conf_mut[sched_ctx], NULL);
  269. starpu_pthread_mutex_init(&hypervisor.resize_mut[sched_ctx], NULL);
  270. _add_config(sched_ctx);
  271. hypervisor.sched_ctx_w[sched_ctx].sched_ctx = sched_ctx;
  272. hypervisor.sched_ctxs[hypervisor.nsched_ctxs++] = sched_ctx;
  273. hypervisor.sched_ctx_w[sched_ctx].total_flops = total_flops;
  274. hypervisor.sched_ctx_w[sched_ctx].remaining_flops = total_flops;
  275. hypervisor.resize[sched_ctx] = 1;
  276. starpu_pthread_mutex_unlock(&act_hypervisor_mutex);
  277. }
  278. static int _get_first_free_sched_ctx(unsigned *sched_ctxs, int nsched_ctxs)
  279. {
  280. int i;
  281. for(i = 0; i < nsched_ctxs; i++)
  282. if(sched_ctxs[i] == STARPU_NMAX_SCHED_CTXS)
  283. return i;
  284. return STARPU_NMAX_SCHED_CTXS;
  285. }
  286. /* rearange array of sched_ctxs in order not to have {MAXVAL, MAXVAL, 5, MAXVAL, 7}
  287. and have instead {5, 7, MAXVAL, MAXVAL, MAXVAL}
  288. it is easier afterwards to iterate the array
  289. */
  290. static void _rearange_sched_ctxs(unsigned *sched_ctxs, int old_nsched_ctxs)
  291. {
  292. int first_free_id = STARPU_NMAX_SCHED_CTXS;
  293. int i;
  294. for(i = 0; i < old_nsched_ctxs; i++)
  295. {
  296. if(sched_ctxs[i] != STARPU_NMAX_SCHED_CTXS)
  297. {
  298. first_free_id = _get_first_free_sched_ctx(sched_ctxs, old_nsched_ctxs);
  299. if(first_free_id != STARPU_NMAX_SCHED_CTXS)
  300. {
  301. sched_ctxs[first_free_id] = sched_ctxs[i];
  302. sched_ctxs[i] = STARPU_NMAX_SCHED_CTXS;
  303. }
  304. }
  305. }
  306. }
  307. /* unregistered contexts will no longer be resized */
  308. void sc_hypervisor_unregister_ctx(unsigned sched_ctx)
  309. {
  310. printf("unregister ctx %d with remaining flops %lf \n", hypervisor.sched_ctx_w[sched_ctx].sched_ctx, hypervisor.sched_ctx_w[sched_ctx].remaining_flops);
  311. if(hypervisor.policy.end_ctx)
  312. hypervisor.policy.end_ctx(sched_ctx);
  313. unsigned father = starpu_sched_ctx_get_inheritor(sched_ctx);
  314. int *pus;
  315. unsigned npus = starpu_sched_ctx_get_workers_list(sched_ctx, &pus);
  316. starpu_sched_ctx_set_priority(pus, npus, father, 1);
  317. free(pus);
  318. starpu_pthread_mutex_lock(&act_hypervisor_mutex);
  319. unsigned i;
  320. for(i = 0; i < hypervisor.nsched_ctxs; i++)
  321. {
  322. if(hypervisor.sched_ctxs[i] == sched_ctx)
  323. {
  324. hypervisor.sched_ctxs[i] = STARPU_NMAX_SCHED_CTXS;
  325. break;
  326. }
  327. }
  328. _rearange_sched_ctxs(hypervisor.sched_ctxs, hypervisor.nsched_ctxs);
  329. hypervisor.nsched_ctxs--;
  330. hypervisor.sched_ctx_w[sched_ctx].sched_ctx = STARPU_NMAX_SCHED_CTXS;
  331. _remove_config(sched_ctx);
  332. starpu_pthread_mutex_destroy(&hypervisor.conf_mut[sched_ctx]);
  333. starpu_pthread_mutex_destroy(&hypervisor.resize_mut[sched_ctx]);
  334. if(hypervisor.nsched_ctxs == 1)
  335. sc_hypervisor_stop_resize(hypervisor.sched_ctxs[0]);
  336. starpu_pthread_mutex_unlock(&act_hypervisor_mutex);
  337. }
  338. double _get_max_speed_gap()
  339. {
  340. return hypervisor.max_speed_gap;
  341. }
  342. unsigned sc_hypervisor_get_resize_criteria()
  343. {
  344. return hypervisor.resize_criteria;
  345. }
  346. static int get_ntasks( int *tasks)
  347. {
  348. int ntasks = 0;
  349. int j;
  350. for(j = 0; j < STARPU_NMAXWORKERS; j++)
  351. {
  352. ntasks += tasks[j];
  353. }
  354. return ntasks;
  355. }
  356. int sc_hypervisor_get_nworkers_ctx(unsigned sched_ctx, enum starpu_worker_archtype arch)
  357. {
  358. int nworkers_ctx = 0;
  359. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sched_ctx);
  360. int worker;
  361. struct starpu_sched_ctx_iterator it;
  362. if(workers->init_iterator)
  363. workers->init_iterator(workers, &it);
  364. while(workers->has_next(workers, &it))
  365. {
  366. worker = workers->get_next(workers, &it);
  367. enum starpu_worker_archtype curr_arch = starpu_worker_get_type(worker);
  368. if(curr_arch == arch || arch == STARPU_ANY_WORKER)
  369. nworkers_ctx++;
  370. }
  371. return nworkers_ctx;
  372. }
  373. static void _set_elapsed_flops_per_sched_ctx(unsigned sched_ctx, double val)
  374. {
  375. int i;
  376. for(i = 0; i < STARPU_NMAXWORKERS; i++)
  377. {
  378. hypervisor.sched_ctx_w[sched_ctx].elapsed_flops[i] = val;
  379. if(val == 0)
  380. {
  381. hypervisor.sched_ctx_w[sched_ctx].elapsed_data[i] = 0;
  382. hypervisor.sched_ctx_w[sched_ctx].elapsed_tasks[i] = 0;
  383. }
  384. }
  385. }
  386. double sc_hypervisor_get_elapsed_flops_per_sched_ctx(struct sc_hypervisor_wrapper* sc_w)
  387. {
  388. double ret_val = 0.0;
  389. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sc_w->sched_ctx);
  390. int worker;
  391. struct starpu_sched_ctx_iterator it;
  392. if(workers->init_iterator)
  393. workers->init_iterator(workers, &it);
  394. while(workers->has_next(workers, &it))
  395. {
  396. worker = workers->get_next(workers, &it);
  397. ret_val += sc_w->elapsed_flops[worker];
  398. }
  399. return ret_val;
  400. }
  401. double sc_hypervisor_get_total_elapsed_flops_per_sched_ctx(struct sc_hypervisor_wrapper* sc_w)
  402. {
  403. double ret_val = 0.0;
  404. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sc_w->sched_ctx);
  405. int worker;
  406. struct starpu_sched_ctx_iterator it;
  407. if(workers->init_iterator)
  408. workers->init_iterator(workers, &it);
  409. while(workers->has_next(workers, &it))
  410. {
  411. worker = workers->get_next(workers, &it);
  412. ret_val += sc_w->total_elapsed_flops[worker];
  413. }
  414. return ret_val;
  415. }
  416. double sc_hypervisor_get_nready_flops_of_all_sons_of_sched_ctx(unsigned sched_ctx)
  417. {
  418. double ready_flops = starpu_sched_ctx_get_nready_flops(sched_ctx);
  419. unsigned *sched_ctxs;
  420. int nsched_ctxs = 0;
  421. sc_hypervisor_get_ctxs_on_level(&sched_ctxs, &nsched_ctxs, starpu_sched_ctx_get_hierarchy_level(sched_ctx), sched_ctx);
  422. int s;
  423. for(s = 0; s < nsched_ctxs; s++)
  424. ready_flops += sc_hypervisor_get_nready_flops_of_all_sons_of_sched_ctx(sched_ctxs[s]);
  425. //ready_flops += starpu_get_nready_flops_of_sched_ctx(sched_ctxs[s]);
  426. return ready_flops;
  427. }
  428. static void _decrement_elapsed_flops_per_worker(unsigned sched_ctx, int worker, double flops)
  429. {
  430. if(starpu_sched_ctx_get_hierarchy_level(sched_ctx) > 0)
  431. {
  432. unsigned father = starpu_sched_ctx_get_inheritor(sched_ctx);
  433. hypervisor.sched_ctx_w[father].elapsed_flops[worker] -= flops;
  434. _decrement_elapsed_flops_per_worker(father, worker, flops);
  435. }
  436. return;
  437. }
  438. void _reset_resize_sample_info(unsigned sender_sched_ctx, unsigned receiver_sched_ctx)
  439. {
  440. double start_time = starpu_timing_now();
  441. if(sender_sched_ctx != STARPU_NMAX_SCHED_CTXS)
  442. {
  443. /* info concerning only the gflops_rate strateg */
  444. struct sc_hypervisor_wrapper *sender_sc_w = &hypervisor.sched_ctx_w[sender_sched_ctx];
  445. sender_sc_w->start_time = start_time;
  446. int i;
  447. for(i = 0; i < STARPU_NMAXWORKERS; i++)
  448. {
  449. sender_sc_w->start_time_w[i] = start_time;
  450. sender_sc_w->idle_time[i] = 0.0;
  451. sender_sc_w->idle_start_time[i] = 0.0;
  452. hypervisor.sched_ctx_w[sender_sched_ctx].exec_time[i] = 0.0;
  453. hypervisor.sched_ctx_w[sender_sched_ctx].exec_start_time[i] = (hypervisor.sched_ctx_w[sender_sched_ctx].exec_start_time[i] != 0.0) ? starpu_timing_now() : 0.0;
  454. _decrement_elapsed_flops_per_worker(sender_sched_ctx, i, hypervisor.sched_ctx_w[sender_sched_ctx].elapsed_flops[i]);
  455. }
  456. _set_elapsed_flops_per_sched_ctx(sender_sched_ctx, 0.0);
  457. }
  458. if(receiver_sched_ctx != STARPU_NMAX_SCHED_CTXS)
  459. {
  460. struct sc_hypervisor_wrapper *receiver_sc_w = &hypervisor.sched_ctx_w[receiver_sched_ctx];
  461. receiver_sc_w->start_time = start_time;
  462. int i;
  463. for(i = 0; i < STARPU_NMAXWORKERS; i++)
  464. {
  465. receiver_sc_w->start_time_w[i] = (receiver_sc_w->start_time_w[i] != 0.0) ? starpu_timing_now() : 0.0;
  466. receiver_sc_w->idle_time[i] = 0.0;
  467. receiver_sc_w->idle_start_time[i] = (receiver_sc_w->exec_start_time[i] != 0.0) ? 0.0 : starpu_timing_now();
  468. hypervisor.sched_ctx_w[receiver_sched_ctx].exec_start_time[i] = (receiver_sc_w->exec_start_time[i] != 0.0) ? starpu_timing_now() : 0.0;
  469. hypervisor.sched_ctx_w[receiver_sched_ctx].exec_time[i] = 0.0;
  470. _decrement_elapsed_flops_per_worker(receiver_sched_ctx, i, hypervisor.sched_ctx_w[receiver_sched_ctx].elapsed_flops[i]);
  471. }
  472. _set_elapsed_flops_per_sched_ctx(receiver_sched_ctx, 0.0);
  473. }
  474. return;
  475. }
  476. /* actually move the workers: the cpus are moved, gpus are only shared */
  477. /* forbids another resize request before this one is take into account */
  478. void sc_hypervisor_move_workers(unsigned sender_sched_ctx, unsigned receiver_sched_ctx, int* workers_to_move, unsigned nworkers_to_move, unsigned now)
  479. {
  480. if(nworkers_to_move > 0 && hypervisor.resize[sender_sched_ctx])
  481. {
  482. _print_current_time();
  483. unsigned j;
  484. printf("resize ctx %d with %d workers", sender_sched_ctx, nworkers_to_move);
  485. for(j = 0; j < nworkers_to_move; j++)
  486. printf(" %d", workers_to_move[j]);
  487. printf("\n");
  488. starpu_fxt_trace_user_event(1);
  489. hypervisor.allow_remove[receiver_sched_ctx] = 0;
  490. starpu_sched_ctx_add_workers(workers_to_move, nworkers_to_move, receiver_sched_ctx);
  491. if(now)
  492. {
  493. unsigned j;
  494. printf("remove now from ctx %d:", sender_sched_ctx);
  495. for(j = 0; j < nworkers_to_move; j++)
  496. printf(" %d", workers_to_move[j]);
  497. printf("\n");
  498. starpu_sched_ctx_remove_workers(workers_to_move, nworkers_to_move, sender_sched_ctx);
  499. hypervisor.allow_remove[receiver_sched_ctx] = 1;
  500. _reset_resize_sample_info(sender_sched_ctx, receiver_sched_ctx);
  501. }
  502. else
  503. {
  504. int ret = starpu_pthread_mutex_trylock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  505. if(ret != EBUSY)
  506. {
  507. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.receiver_sched_ctx = receiver_sched_ctx;
  508. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.moved_workers = (int*)malloc(nworkers_to_move * sizeof(int));
  509. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.nmoved_workers = nworkers_to_move;
  510. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.acked_workers = (int*)malloc(nworkers_to_move * sizeof(int));
  511. unsigned i;
  512. for(i = 0; i < nworkers_to_move; i++)
  513. {
  514. hypervisor.sched_ctx_w[sender_sched_ctx].current_idle_time[workers_to_move[i]] = 0.0;
  515. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.moved_workers[i] = workers_to_move[i];
  516. hypervisor.sched_ctx_w[sender_sched_ctx].resize_ack.acked_workers[i] = 0;
  517. }
  518. hypervisor.resize[sender_sched_ctx] = 0;
  519. if(imposed_resize) imposed_resize = 0;
  520. starpu_pthread_mutex_unlock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  521. }
  522. }
  523. struct sc_hypervisor_policy_config *new_config = sc_hypervisor_get_config(receiver_sched_ctx);
  524. unsigned i;
  525. for(i = 0; i < nworkers_to_move; i++)
  526. new_config->max_idle[workers_to_move[i]] = new_config->max_idle[workers_to_move[i]] !=MAX_IDLE_TIME ? new_config->max_idle[workers_to_move[i]] : new_config->new_workers_max_idle;
  527. }
  528. return;
  529. }
  530. void sc_hypervisor_add_workers_to_sched_ctx(int* workers_to_add, unsigned nworkers_to_add, unsigned sched_ctx)
  531. {
  532. if(nworkers_to_add > 0 && hypervisor.resize[sched_ctx])
  533. {
  534. _print_current_time();
  535. unsigned j;
  536. printf("add to ctx %d:", sched_ctx);
  537. for(j = 0; j < nworkers_to_add; j++)
  538. printf(" %d", workers_to_add[j]);
  539. printf("\n");
  540. starpu_sched_ctx_add_workers(workers_to_add, nworkers_to_add, sched_ctx);
  541. struct sc_hypervisor_policy_config *new_config = sc_hypervisor_get_config(sched_ctx);
  542. unsigned i;
  543. for(i = 0; i < nworkers_to_add; i++)
  544. new_config->max_idle[workers_to_add[i]] = new_config->max_idle[workers_to_add[i]] != MAX_IDLE_TIME ? new_config->max_idle[workers_to_add[i]] : new_config->new_workers_max_idle;
  545. _reset_resize_sample_info(STARPU_NMAX_SCHED_CTXS, sched_ctx);
  546. }
  547. return;
  548. }
  549. unsigned sc_hypervisor_can_resize(unsigned sched_ctx)
  550. {
  551. return hypervisor.resize[sched_ctx];
  552. }
  553. void sc_hypervisor_remove_workers_from_sched_ctx(int* workers_to_remove, unsigned nworkers_to_remove, unsigned sched_ctx, unsigned now)
  554. {
  555. if(nworkers_to_remove > 0 && hypervisor.resize[sched_ctx] && hypervisor.allow_remove[sched_ctx])
  556. {
  557. _print_current_time();
  558. unsigned nworkers = 0;
  559. int workers[nworkers_to_remove];
  560. if(now)
  561. {
  562. unsigned j;
  563. printf("remove explicitley now from ctx %d:", sched_ctx);
  564. for(j = 0; j < nworkers_to_remove; j++)
  565. printf(" %d", workers_to_remove[j]);
  566. printf("\n");
  567. starpu_sched_ctx_remove_workers(workers_to_remove, nworkers_to_remove, sched_ctx);
  568. _reset_resize_sample_info(sched_ctx, STARPU_NMAX_SCHED_CTXS);
  569. }
  570. else
  571. {
  572. printf("try to remove from ctx %d: ", sched_ctx);
  573. unsigned j;
  574. for(j = 0; j < nworkers_to_remove; j++)
  575. printf(" %d", workers_to_remove[j]);
  576. printf("\n");
  577. int ret = starpu_pthread_mutex_trylock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  578. if(ret != EBUSY)
  579. {
  580. unsigned i;
  581. for(i = 0; i < nworkers_to_remove; i++)
  582. if(starpu_sched_ctx_contains_worker(workers_to_remove[i], sched_ctx))
  583. workers[nworkers++] = workers_to_remove[i];
  584. hypervisor.sched_ctx_w[sched_ctx].resize_ack.receiver_sched_ctx = -1;
  585. hypervisor.sched_ctx_w[sched_ctx].resize_ack.moved_workers = (int*)malloc(nworkers_to_remove * sizeof(int));
  586. hypervisor.sched_ctx_w[sched_ctx].resize_ack.nmoved_workers = (int)nworkers;
  587. hypervisor.sched_ctx_w[sched_ctx].resize_ack.acked_workers = (int*)malloc(nworkers_to_remove * sizeof(int));
  588. for(i = 0; i < nworkers; i++)
  589. {
  590. hypervisor.sched_ctx_w[sched_ctx].current_idle_time[workers[i]] = 0.0;
  591. hypervisor.sched_ctx_w[sched_ctx].resize_ack.moved_workers[i] = workers[i];
  592. hypervisor.sched_ctx_w[sched_ctx].resize_ack.acked_workers[i] = 0;
  593. }
  594. hypervisor.resize[sched_ctx] = 0;
  595. if(imposed_resize) imposed_resize = 0;
  596. starpu_pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  597. }
  598. }
  599. }
  600. return;
  601. }
  602. static unsigned _ack_resize_completed(unsigned sched_ctx, int worker)
  603. {
  604. if(worker != -1 && !starpu_sched_ctx_contains_worker(worker, sched_ctx))
  605. return 0;
  606. struct sc_hypervisor_resize_ack *resize_ack = NULL;
  607. unsigned sender_sched_ctx = STARPU_NMAX_SCHED_CTXS;
  608. int i;
  609. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  610. {
  611. if(hypervisor.sched_ctxs[i] != STARPU_NMAX_SCHED_CTXS)
  612. {
  613. struct sc_hypervisor_wrapper *sc_w = &hypervisor.sched_ctx_w[hypervisor.sched_ctxs[i]];
  614. starpu_pthread_mutex_lock(&sc_w->mutex);
  615. unsigned only_remove = 0;
  616. if(sc_w->resize_ack.receiver_sched_ctx == -1 && hypervisor.sched_ctxs[i] != sched_ctx &&
  617. sc_w->resize_ack.nmoved_workers > 0 && starpu_sched_ctx_contains_worker(worker, hypervisor.sched_ctxs[i]))
  618. {
  619. int j;
  620. for(j = 0; j < sc_w->resize_ack.nmoved_workers; j++)
  621. if(sc_w->resize_ack.moved_workers[j] == worker)
  622. {
  623. only_remove = 1;
  624. _reset_resize_sample_info(sched_ctx, STARPU_NMAX_SCHED_CTXS);
  625. starpu_pthread_mutex_unlock(&sc_w->mutex);
  626. break;
  627. }
  628. }
  629. if(only_remove ||
  630. (sc_w->resize_ack.receiver_sched_ctx != -1 && sc_w->resize_ack.receiver_sched_ctx == (int)sched_ctx))
  631. {
  632. resize_ack = &sc_w->resize_ack;
  633. sender_sched_ctx = hypervisor.sched_ctxs[i];
  634. starpu_pthread_mutex_unlock(&sc_w->mutex);
  635. break;
  636. }
  637. starpu_pthread_mutex_unlock(&sc_w->mutex);
  638. }
  639. }
  640. /* if there is no ctx waiting for its ack return 1*/
  641. if(resize_ack == NULL)
  642. {
  643. return 1;
  644. }
  645. int ret = starpu_pthread_mutex_trylock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  646. if(ret != EBUSY)
  647. {
  648. int *moved_workers = resize_ack->moved_workers;
  649. int nmoved_workers = resize_ack->nmoved_workers;
  650. int *acked_workers = resize_ack->acked_workers;
  651. if(worker != -1)
  652. {
  653. for(i = 0; i < nmoved_workers; i++)
  654. {
  655. int moved_worker = moved_workers[i];
  656. if(moved_worker == worker && acked_workers[i] == 0)
  657. {
  658. acked_workers[i] = 1;
  659. }
  660. }
  661. }
  662. int nacked_workers = 0;
  663. for(i = 0; i < nmoved_workers; i++)
  664. {
  665. nacked_workers += (acked_workers[i] == 1);
  666. }
  667. unsigned resize_completed = (nacked_workers == nmoved_workers);
  668. int receiver_sched_ctx = sched_ctx;
  669. if(resize_completed)
  670. {
  671. /* if the permission to resize is not allowed by the user don't do it
  672. whatever the application says */
  673. if(!((hypervisor.resize[sender_sched_ctx] == 0 || hypervisor.resize[receiver_sched_ctx] == 0) && imposed_resize))
  674. {
  675. /* int j; */
  676. /* printf("remove after ack from ctx %d:", sender_sched_ctx); */
  677. /* for(j = 0; j < nmoved_workers; j++) */
  678. /* printf(" %d", moved_workers[j]); */
  679. /* printf("\n"); */
  680. starpu_sched_ctx_remove_workers(moved_workers, nmoved_workers, sender_sched_ctx);
  681. _reset_resize_sample_info(sender_sched_ctx, receiver_sched_ctx);
  682. hypervisor.resize[sender_sched_ctx] = 1;
  683. hypervisor.allow_remove[receiver_sched_ctx] = 1;
  684. /* if the user allowed resizing leave the decisions to the application */
  685. if(imposed_resize) imposed_resize = 0;
  686. resize_ack->receiver_sched_ctx = -1;
  687. resize_ack->nmoved_workers = 0;
  688. free(resize_ack->moved_workers);
  689. free(resize_ack->acked_workers);
  690. }
  691. starpu_pthread_mutex_unlock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  692. return resize_completed;
  693. }
  694. starpu_pthread_mutex_unlock(&hypervisor.sched_ctx_w[sender_sched_ctx].mutex);
  695. }
  696. return 0;
  697. }
  698. /* Enqueue a resize request for 'sched_ctx', to be executed when the
  699. * 'task_tag' tasks of 'sched_ctx' complete. */
  700. void sc_hypervisor_post_resize_request(unsigned sched_ctx, int task_tag)
  701. {
  702. struct resize_request_entry *entry;
  703. entry = malloc(sizeof *entry);
  704. STARPU_ASSERT(entry != NULL);
  705. entry->sched_ctx = sched_ctx;
  706. entry->task_tag = task_tag;
  707. starpu_pthread_mutex_lock(&hypervisor.resize_mut[sched_ctx]);
  708. HASH_ADD_INT(hypervisor.resize_requests[sched_ctx], task_tag, entry);
  709. starpu_pthread_mutex_unlock(&hypervisor.resize_mut[sched_ctx]);
  710. }
  711. void sc_hypervisor_resize_ctxs(unsigned *sched_ctxs, int nsched_ctxs , int *workers, int nworkers)
  712. {
  713. if(hypervisor.policy.resize_ctxs)
  714. hypervisor.policy.resize_ctxs(sched_ctxs, nsched_ctxs, workers, nworkers);
  715. }
  716. void _sc_hypervisor_allow_compute_idle(unsigned sched_ctx, int worker, unsigned allow)
  717. {
  718. hypervisor.sched_ctx_w[sched_ctx].compute_idle[worker] = allow;
  719. }
  720. void sc_hypervisor_update_resize_interval(unsigned *sched_ctxs, int nsched_ctxs)
  721. {
  722. unsigned sched_ctx;
  723. int total_max_nworkers = 0;
  724. int max_cpus = starpu_cpu_worker_get_count();
  725. unsigned configured = 0;
  726. int i;
  727. for(i = 0; i < nsched_ctxs; i++)
  728. {
  729. sched_ctx = sched_ctxs[i];
  730. if(hypervisor.sched_ctx_w[sched_ctx].to_be_sized) continue;
  731. struct sc_hypervisor_policy_config *config = sc_hypervisor_get_config(sched_ctx);
  732. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sched_ctx);
  733. int worker;
  734. struct starpu_sched_ctx_iterator it;
  735. if(workers->init_iterator)
  736. workers->init_iterator(workers, &it);
  737. double elapsed_time_worker[STARPU_NMAXWORKERS];
  738. double norm_idle_time = 0.0;
  739. double end_time = starpu_timing_now();
  740. while(workers->has_next(workers, &it))
  741. {
  742. double idle_time = 0.0;
  743. worker = workers->get_next(workers, &it);
  744. if(hypervisor.sched_ctx_w[sched_ctx].compute_idle[worker])
  745. {
  746. if(hypervisor.sched_ctx_w[sched_ctx].start_time_w[worker] == 0.0)
  747. elapsed_time_worker[worker] = 0.0;
  748. else
  749. elapsed_time_worker[worker] = (end_time - hypervisor.sched_ctx_w[sched_ctx].start_time_w[worker]) / 1000000.0;
  750. if(hypervisor.sched_ctx_w[sched_ctx].idle_start_time[worker] == 0.0)
  751. {
  752. idle_time = hypervisor.sched_ctx_w[sched_ctx].idle_time[worker]; /* in seconds */
  753. }
  754. else
  755. {
  756. double idle = (end_time - hypervisor.sched_ctx_w[sched_ctx].idle_start_time[worker]) / 1000000.0; /* in seconds */
  757. idle_time = hypervisor.sched_ctx_w[sched_ctx].idle_time[worker] + idle;
  758. }
  759. norm_idle_time += (elapsed_time_worker[worker] == 0.0 ? 0.0 : (idle_time / elapsed_time_worker[worker]));
  760. /* printf("%d/%d: start time %lf elapsed time %lf idle time %lf norm_idle_time %lf \n", */
  761. /* worker, sched_ctx, hypervisor.sched_ctx_w[sched_ctx].start_time_w[worker], elapsed_time_worker[worker], idle_time, norm_idle_time); */
  762. }
  763. }
  764. double norm_exec_time = 0.0;
  765. for(worker = 0; worker < STARPU_NMAXWORKERS; worker++)
  766. {
  767. double exec_time = 0.0;
  768. if(hypervisor.sched_ctx_w[sched_ctx].start_time_w[worker] == 0.0)
  769. elapsed_time_worker[worker] = 0.0;
  770. else
  771. elapsed_time_worker[worker] = (end_time - hypervisor.sched_ctx_w[sched_ctx].start_time_w[worker]) / 1000000.0;
  772. if(hypervisor.sched_ctx_w[sched_ctx].exec_start_time[worker] == 0.0)
  773. {
  774. exec_time = hypervisor.sched_ctx_w[sched_ctx].exec_time[worker];
  775. }
  776. else
  777. {
  778. double current_exec_time = (end_time - hypervisor.sched_ctx_w[sched_ctx].exec_start_time[worker]) / 1000000.0; /* in seconds */
  779. exec_time = hypervisor.sched_ctx_w[sched_ctx].exec_time[worker] + current_exec_time;
  780. }
  781. norm_exec_time += elapsed_time_worker[worker] == 0.0 ? 0.0 : exec_time / elapsed_time_worker[worker];
  782. }
  783. double curr_time = starpu_timing_now();
  784. double elapsed_time = (curr_time - hypervisor.sched_ctx_w[sched_ctx].start_time) / 1000000.0; /* in seconds */
  785. int nready_tasks = starpu_sched_ctx_get_nready_tasks(sched_ctx);
  786. if(norm_idle_time >= 0.9)
  787. {
  788. config->max_nworkers = lrint(norm_exec_time);
  789. }
  790. else
  791. {
  792. if(norm_idle_time < 0.1)
  793. config->max_nworkers = lrint(norm_exec_time) + nready_tasks - 1; //workers->nworkers + hypervisor.sched_ctx_w[sched_ctx].nready_tasks - 1;
  794. else
  795. config->max_nworkers = lrint(norm_exec_time);
  796. }
  797. // config->max_nworkers = hypervisor.sched_ctx_w[sched_ctx].nready_tasks - 1;
  798. if(config->max_nworkers < 0)
  799. config->max_nworkers = 0;
  800. if(config->max_nworkers > max_cpus)
  801. config->max_nworkers = max_cpus;
  802. printf("%d: ready tasks %d norm_idle_time %lf elapsed_time %lf norm_exec_time %lf nworker %d max %d \n",
  803. sched_ctx, nready_tasks, norm_idle_time, elapsed_time, norm_exec_time, workers->nworkers, config->max_nworkers);
  804. total_max_nworkers += config->max_nworkers;
  805. configured = 1;
  806. }
  807. /*if the sum of the max cpus is smaller than the total cpus available
  808. increase the max for the ones having more ready tasks to exec */
  809. if(configured && total_max_nworkers < max_cpus)
  810. {
  811. int diff = max_cpus - total_max_nworkers;
  812. int max_nready = -1;
  813. unsigned max_nready_sched_ctx = sched_ctxs[0];
  814. for(i = 0; i < nsched_ctxs; i++)
  815. {
  816. int nready_tasks = starpu_sched_ctx_get_nready_tasks(sched_ctxs[i]);
  817. if(max_nready < nready_tasks)
  818. {
  819. max_nready = nready_tasks;
  820. max_nready_sched_ctx = sched_ctxs[i];
  821. }
  822. }
  823. struct sc_hypervisor_policy_config *config = sc_hypervisor_get_config(max_nready_sched_ctx);
  824. config->max_nworkers += diff;
  825. printf("%d: redib max_nworkers incr %d \n", max_nready_sched_ctx, config->max_nworkers);
  826. }
  827. }
  828. /* notifies the hypervisor that a new task was pushed on the queue of the worker */
  829. static void notify_pushed_task(unsigned sched_ctx, int worker)
  830. {
  831. hypervisor.sched_ctx_w[sched_ctx].pushed_tasks[worker]++;
  832. if(hypervisor.sched_ctx_w[sched_ctx].total_flops != 0.0 && hypervisor.sched_ctx_w[sched_ctx].start_time == 0.0)
  833. hypervisor.sched_ctx_w[sched_ctx].start_time = starpu_timing_now();
  834. if(hypervisor.sched_ctx_w[sched_ctx].total_flops != 0.0 && hypervisor.sched_ctx_w[sched_ctx].start_time_w[worker] == 0.0)
  835. hypervisor.sched_ctx_w[sched_ctx].start_time_w[worker] = starpu_timing_now();
  836. if(hypervisor.sched_ctx_w[sched_ctx].total_flops != 0.0 && hypervisor.sched_ctx_w[sched_ctx].real_start_time == 0.0)
  837. hypervisor.sched_ctx_w[sched_ctx].real_start_time = starpu_timing_now();
  838. int ntasks = get_ntasks(hypervisor.sched_ctx_w[sched_ctx].pushed_tasks);
  839. if((hypervisor.min_tasks == 0 || (!(hypervisor.resize[sched_ctx] == 0 && imposed_resize) && ntasks == hypervisor.min_tasks)) && hypervisor.check_min_tasks[sched_ctx])
  840. {
  841. hypervisor.resize[sched_ctx] = 1;
  842. if(imposed_resize) imposed_resize = 0;
  843. hypervisor.check_min_tasks[sched_ctx] = 0;
  844. }
  845. if(hypervisor.policy.handle_pushed_task)
  846. hypervisor.policy.handle_pushed_task(sched_ctx, worker);
  847. }
  848. unsigned choose_ctx_to_steal(int worker)
  849. {
  850. int j;
  851. int ns = hypervisor.nsched_ctxs;
  852. int max_ready_tasks = 0;
  853. unsigned chosen_ctx = STARPU_NMAX_SCHED_CTXS;
  854. for(j = 0; j < ns; j++)
  855. {
  856. unsigned other_ctx = hypervisor.sched_ctxs[j];
  857. int nready = starpu_sched_ctx_get_nready_tasks(other_ctx);
  858. if(!starpu_sched_ctx_contains_worker(worker, other_ctx) && max_ready_tasks < nready)
  859. {
  860. max_ready_tasks = nready;
  861. chosen_ctx = other_ctx;
  862. }
  863. }
  864. return chosen_ctx;
  865. }
  866. /* notifies the hypervisor that the worker spent another cycle in idle time */
  867. static void notify_idle_cycle(unsigned sched_ctx, int worker, double idle_time)
  868. {
  869. struct sc_hypervisor_wrapper *sc_w = &hypervisor.sched_ctx_w[sched_ctx];
  870. sc_w->current_idle_time[worker] += idle_time;
  871. if(sc_w->idle_start_time[worker] == 0.0 && sc_w->hyp_react_start_time != 0.0)
  872. sc_w->idle_start_time[worker] = starpu_timing_now();
  873. if(sc_w->idle_start_time[worker] > 0.0)
  874. {
  875. double end_time = starpu_timing_now();
  876. sc_w->idle_time[worker] += (end_time - sc_w->idle_start_time[worker]) / 1000000.0; /* in seconds */
  877. }
  878. hypervisor.sched_ctx_w[sched_ctx].idle_start_time[worker] = starpu_timing_now();
  879. if(hypervisor.resize[sched_ctx] && hypervisor.policy.handle_idle_cycle)
  880. {
  881. if(sc_w->sched_ctx != STARPU_NMAX_SCHED_CTXS && sc_w->hyp_react_start_time != 0.0)
  882. {
  883. unsigned idle_everywhere = 0;
  884. int ret = starpu_pthread_mutex_trylock(&act_hypervisor_mutex);
  885. if(ret != EBUSY)
  886. {
  887. if(sc_hypervisor_check_idle(sched_ctx, worker))
  888. {
  889. idle_everywhere = 1;
  890. unsigned *sched_ctxs = NULL;
  891. unsigned nsched_ctxs = starpu_worker_get_sched_ctx_list(worker, &sched_ctxs);
  892. int s;
  893. for(s = 0; s < nsched_ctxs; s++)
  894. {
  895. if(hypervisor.sched_ctx_w[sched_ctxs[s]].sched_ctx != STARPU_NMAX_SCHED_CTXS)
  896. {
  897. if(!sc_hypervisor_check_idle(sched_ctxs[s], worker))
  898. idle_everywhere = 0;
  899. }
  900. }
  901. free(sched_ctxs);
  902. }
  903. starpu_pthread_mutex_unlock(&act_hypervisor_mutex);
  904. }
  905. if(idle_everywhere)
  906. hypervisor.policy.handle_idle_cycle(sched_ctx, worker);
  907. }
  908. }
  909. return;
  910. }
  911. /* notifies the hypervisor that the worker is no longer idle and a new task was pushed on its queue */
  912. static void notify_poped_task(unsigned sched_ctx, int worker)
  913. {
  914. if(hypervisor.sched_ctx_w[sched_ctx].start_time_w[worker] == 0.0)
  915. hypervisor.sched_ctx_w[sched_ctx].start_time_w[worker] = starpu_timing_now();
  916. hypervisor.sched_ctx_w[sched_ctx].exec_start_time[worker] = starpu_timing_now();
  917. if(hypervisor.sched_ctx_w[sched_ctx].idle_start_time[worker] > 0.0)
  918. {
  919. int ns = hypervisor.nsched_ctxs;
  920. int j;
  921. for(j = 0; j < ns; j++)
  922. {
  923. if(hypervisor.sched_ctxs[j] != sched_ctx)
  924. {
  925. if(hypervisor.sched_ctx_w[hypervisor.sched_ctxs[j]].idle_start_time[worker] > 0.0)
  926. hypervisor.sched_ctx_w[hypervisor.sched_ctxs[j]].compute_partial_idle[worker] = 1;
  927. }
  928. }
  929. double end_time = starpu_timing_now();
  930. double idle = (end_time - hypervisor.sched_ctx_w[sched_ctx].idle_start_time[worker]) / 1000000.0; /* in seconds */
  931. if(hypervisor.sched_ctx_w[sched_ctx].compute_partial_idle[worker])
  932. hypervisor.sched_ctx_w[sched_ctx].idle_time[worker] += idle / 2.0;
  933. else
  934. hypervisor.sched_ctx_w[sched_ctx].idle_time[worker] += idle;
  935. hypervisor.sched_ctx_w[sched_ctx].compute_partial_idle[worker] = 0;
  936. hypervisor.sched_ctx_w[sched_ctx].idle_start_time[worker] = 0.0;
  937. }
  938. if(hypervisor.resize[sched_ctx])
  939. hypervisor.sched_ctx_w[sched_ctx].current_idle_time[worker] = 0.0;
  940. if(hypervisor.policy.handle_idle_end)
  941. hypervisor.policy.handle_idle_end(sched_ctx, worker);
  942. }
  943. static void _update_counters_hierarchically(int worker, unsigned sched_ctx, double flops, size_t data_size)
  944. {
  945. hypervisor.sched_ctx_w[sched_ctx].poped_tasks[worker]++;
  946. hypervisor.sched_ctx_w[sched_ctx].elapsed_flops[worker] += flops;
  947. hypervisor.sched_ctx_w[sched_ctx].elapsed_data[worker] += data_size ;
  948. hypervisor.sched_ctx_w[sched_ctx].elapsed_tasks[worker]++ ;
  949. hypervisor.sched_ctx_w[sched_ctx].total_elapsed_flops[worker] += flops;
  950. starpu_pthread_mutex_lock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  951. hypervisor.sched_ctx_w[sched_ctx].remaining_flops -= flops;
  952. starpu_pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  953. if(starpu_sched_ctx_get_hierarchy_level(sched_ctx) > 0)
  954. _update_counters_hierarchically(worker, starpu_sched_ctx_get_inheritor(sched_ctx), flops, data_size);
  955. return;
  956. }
  957. /* notifies the hypervisor that a tagged task has just been executed */
  958. static void notify_post_exec_task(struct starpu_task *task, size_t data_size, uint32_t footprint, int task_tag, double flops)
  959. {
  960. unsigned sched_ctx = task->sched_ctx;
  961. int worker = starpu_worker_get_id();
  962. if(hypervisor.sched_ctx_w[sched_ctx].exec_start_time[worker] != 0.0)
  963. {
  964. double current_time = starpu_timing_now();
  965. hypervisor.sched_ctx_w[sched_ctx].exec_time[worker] += (current_time -
  966. hypervisor.sched_ctx_w[sched_ctx].exec_start_time[worker]) / 1000000.0; /* in seconds */
  967. hypervisor.sched_ctx_w[sched_ctx].exec_start_time[worker] = 0.0;
  968. }
  969. hypervisor.sched_ctx_w[sched_ctx].poped_tasks[worker]++;
  970. hypervisor.sched_ctx_w[sched_ctx].elapsed_flops[worker] += flops;
  971. hypervisor.sched_ctx_w[sched_ctx].elapsed_data[worker] += data_size ;
  972. hypervisor.sched_ctx_w[sched_ctx].elapsed_tasks[worker]++ ;
  973. hypervisor.sched_ctx_w[sched_ctx].total_elapsed_flops[worker] += flops;
  974. starpu_pthread_mutex_lock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  975. hypervisor.sched_ctx_w[sched_ctx].remaining_flops -= flops;
  976. if(_sc_hypervisor_use_lazy_resize())
  977. _ack_resize_completed(sched_ctx, worker);
  978. starpu_pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  979. if(starpu_sched_ctx_get_hierarchy_level(sched_ctx) > 0)
  980. {
  981. _update_counters_hierarchically(worker, starpu_sched_ctx_get_inheritor(sched_ctx), flops, data_size);
  982. }
  983. if(hypervisor.resize[sched_ctx])
  984. {
  985. if(hypervisor.policy.handle_poped_task)
  986. {
  987. if(hypervisor.sched_ctx_w[sched_ctx].hyp_react_start_time == 0.0)
  988. hypervisor.sched_ctx_w[sched_ctx].hyp_react_start_time = starpu_timing_now();
  989. double curr_time = starpu_timing_now();
  990. double elapsed_time = (curr_time - hypervisor.sched_ctx_w[sched_ctx].hyp_react_start_time) / 1000000.0; /* in seconds */
  991. if(hypervisor.sched_ctx_w[sched_ctx].sched_ctx != STARPU_NMAX_SCHED_CTXS && elapsed_time > hypervisor.sched_ctx_w[sched_ctx].config->time_sample)
  992. {
  993. hypervisor.policy.handle_poped_task(sched_ctx, worker, task, footprint);
  994. hypervisor.sched_ctx_w[sched_ctx].hyp_react_start_time = starpu_timing_now();
  995. }
  996. }
  997. }
  998. /* starpu_pthread_mutex_lock(&act_hypervisor_mutex); */
  999. /* _ack_resize_completed(sched_ctx, worker); */
  1000. /* starpu_pthread_mutex_unlock(&act_hypervisor_mutex); */
  1001. if(hypervisor.sched_ctx_w[sched_ctx].poped_tasks[worker] % 200 == 0)
  1002. _print_current_time();
  1003. if(task_tag <= 0)
  1004. return;
  1005. unsigned conf_sched_ctx;
  1006. unsigned i;
  1007. unsigned ns = hypervisor.nsched_ctxs;
  1008. for(i = 0; i < ns; i++)
  1009. {
  1010. struct configuration_entry *entry;
  1011. conf_sched_ctx = hypervisor.sched_ctxs[i];
  1012. starpu_pthread_mutex_lock(&hypervisor.conf_mut[conf_sched_ctx]);
  1013. HASH_FIND_INT(hypervisor.configurations[conf_sched_ctx], &task_tag, entry);
  1014. if (entry != NULL)
  1015. {
  1016. struct sc_hypervisor_policy_config *config = entry->configuration;
  1017. sc_hypervisor_set_config(conf_sched_ctx, config);
  1018. HASH_DEL(hypervisor.configurations[conf_sched_ctx], entry);
  1019. free(config);
  1020. }
  1021. starpu_pthread_mutex_unlock(&hypervisor.conf_mut[conf_sched_ctx]);
  1022. }
  1023. if(hypervisor.resize[sched_ctx])
  1024. {
  1025. starpu_pthread_mutex_lock(&hypervisor.resize_mut[sched_ctx]);
  1026. if(hypervisor.policy.handle_post_exec_hook)
  1027. {
  1028. /* Check whether 'task_tag' is in the 'resize_requests' set. */
  1029. struct resize_request_entry *entry;
  1030. HASH_FIND_INT(hypervisor.resize_requests[sched_ctx], &task_tag, entry);
  1031. if (entry != NULL)
  1032. {
  1033. hypervisor.policy.handle_post_exec_hook(sched_ctx, task_tag);
  1034. HASH_DEL(hypervisor.resize_requests[sched_ctx], entry);
  1035. free(entry);
  1036. }
  1037. }
  1038. starpu_pthread_mutex_unlock(&hypervisor.resize_mut[sched_ctx]);
  1039. }
  1040. return;
  1041. }
  1042. static void notify_submitted_job(struct starpu_task *task, uint32_t footprint, size_t data_size)
  1043. {
  1044. unsigned sched_ctx = task->sched_ctx;
  1045. starpu_pthread_mutex_lock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  1046. hypervisor.sched_ctx_w[sched_ctx].submitted_flops += task->flops;
  1047. starpu_pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  1048. if(hypervisor.policy.handle_submitted_job && !type_of_tasks_known)
  1049. hypervisor.policy.handle_submitted_job(task->cl, task->sched_ctx, footprint, data_size);
  1050. }
  1051. static void notify_empty_ctx(unsigned sched_ctx_id, struct starpu_task *task)
  1052. {
  1053. sc_hypervisor_resize_ctxs(NULL, -1 , NULL, -1);
  1054. }
  1055. void sc_hypervisor_set_type_of_task(struct starpu_codelet *cl, unsigned sched_ctx, uint32_t footprint, size_t data_size)
  1056. {
  1057. type_of_tasks_known = 1;
  1058. if(hypervisor.policy.handle_submitted_job)
  1059. hypervisor.policy.handle_submitted_job(cl, sched_ctx, footprint, data_size);
  1060. }
  1061. static void notify_delete_context(unsigned sched_ctx)
  1062. {
  1063. _print_current_time();
  1064. sc_hypervisor_unregister_ctx(sched_ctx);
  1065. }
  1066. void sc_hypervisor_size_ctxs(unsigned *sched_ctxs, int nsched_ctxs, int *workers, int nworkers)
  1067. {
  1068. // starpu_pthread_mutex_lock(&act_hypervisor_mutex);
  1069. unsigned curr_nsched_ctxs = sched_ctxs == NULL ? hypervisor.nsched_ctxs : (unsigned)nsched_ctxs;
  1070. unsigned *curr_sched_ctxs = sched_ctxs == NULL ? hypervisor.sched_ctxs : sched_ctxs;
  1071. // starpu_pthread_mutex_unlock(&act_hypervisor_mutex);
  1072. unsigned s;
  1073. for(s = 0; s < curr_nsched_ctxs; s++)
  1074. hypervisor.resize[curr_sched_ctxs[s]] = 1;
  1075. if(hypervisor.policy.size_ctxs)
  1076. hypervisor.policy.size_ctxs(curr_sched_ctxs, curr_nsched_ctxs, workers, nworkers);
  1077. }
  1078. struct sc_hypervisor_wrapper* sc_hypervisor_get_wrapper(unsigned sched_ctx)
  1079. {
  1080. return &hypervisor.sched_ctx_w[sched_ctx];
  1081. }
  1082. unsigned* sc_hypervisor_get_sched_ctxs()
  1083. {
  1084. return hypervisor.sched_ctxs;
  1085. }
  1086. int sc_hypervisor_get_nsched_ctxs()
  1087. {
  1088. int ns;
  1089. ns = hypervisor.nsched_ctxs;
  1090. return ns;
  1091. }
  1092. int _sc_hypervisor_use_lazy_resize(void)
  1093. {
  1094. char* lazy = getenv("SC_HYPERVISOR_LAZY_RESIZE");
  1095. return lazy ? atoi(lazy) : 1;
  1096. }
  1097. void sc_hypervisor_save_size_req(unsigned *sched_ctxs, int nsched_ctxs, int *workers, int nworkers)
  1098. {
  1099. hypervisor.sr = (struct size_request*)malloc(sizeof(struct size_request));
  1100. hypervisor.sr->sched_ctxs = sched_ctxs;
  1101. hypervisor.sr->nsched_ctxs = nsched_ctxs;
  1102. hypervisor.sr->workers = workers;
  1103. hypervisor.sr->nworkers = nworkers;
  1104. }
  1105. unsigned sc_hypervisor_get_size_req(unsigned **sched_ctxs, int* nsched_ctxs, int **workers, int *nworkers)
  1106. {
  1107. if(hypervisor.sr != NULL)
  1108. {
  1109. *sched_ctxs = hypervisor.sr->sched_ctxs;
  1110. *nsched_ctxs = hypervisor.sr->nsched_ctxs;
  1111. *workers = hypervisor.sr->workers;
  1112. *nworkers = hypervisor.sr->nworkers;
  1113. return 1;
  1114. }
  1115. return 0;
  1116. }
  1117. void sc_hypervisor_free_size_req(void)
  1118. {
  1119. if(hypervisor.sr != NULL)
  1120. {
  1121. free(hypervisor.sr);
  1122. hypervisor.sr = NULL;
  1123. }
  1124. }
  1125. double _get_optimal_v(unsigned sched_ctx)
  1126. {
  1127. return hypervisor.optimal_v[sched_ctx];
  1128. }
  1129. void _set_optimal_v(unsigned sched_ctx, double optimal_v)
  1130. {
  1131. hypervisor.optimal_v[sched_ctx] = optimal_v;
  1132. }
  1133. static struct types_of_workers* _init_structure_types_of_workers(void)
  1134. {
  1135. struct types_of_workers *tw = (struct types_of_workers*)malloc(sizeof(struct types_of_workers));
  1136. tw->ncpus = 0;
  1137. tw->ncuda = 0;
  1138. tw->nw = 0;
  1139. return tw;
  1140. }
  1141. struct types_of_workers* sc_hypervisor_get_types_of_workers(int *workers, unsigned nworkers)
  1142. {
  1143. struct types_of_workers *tw = _init_structure_types_of_workers();
  1144. unsigned w;
  1145. for(w = 0; w < nworkers; w++)
  1146. {
  1147. enum starpu_worker_archtype arch = workers == NULL ? starpu_worker_get_type((int)w) : starpu_worker_get_type(workers[w]);
  1148. if(arch == STARPU_CPU_WORKER)
  1149. tw->ncpus++;
  1150. if(arch == STARPU_CUDA_WORKER)
  1151. tw->ncuda++;
  1152. }
  1153. if(tw->ncpus > 0) tw->nw++;
  1154. if(tw->ncuda > 0) tw->nw++;
  1155. return tw;
  1156. }
  1157. void sc_hypervisor_update_diff_total_flops(unsigned sched_ctx, double diff_total_flops)
  1158. {
  1159. starpu_pthread_mutex_lock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  1160. hypervisor.sched_ctx_w[sched_ctx].total_flops += diff_total_flops;
  1161. hypervisor.sched_ctx_w[sched_ctx].remaining_flops += diff_total_flops;
  1162. starpu_pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  1163. if(starpu_sched_ctx_get_hierarchy_level(sched_ctx) > 0)
  1164. sc_hypervisor_update_diff_total_flops(starpu_sched_ctx_get_inheritor(sched_ctx), diff_total_flops);
  1165. }
  1166. void sc_hypervisor_update_diff_elapsed_flops(unsigned sched_ctx, double diff_elapsed_flops)
  1167. {
  1168. int workerid = starpu_worker_get_id();
  1169. if(workerid != -1)
  1170. {
  1171. // starpu_pthread_mutex_lock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  1172. hypervisor.sched_ctx_w[sched_ctx].elapsed_flops[workerid] += diff_elapsed_flops;
  1173. hypervisor.sched_ctx_w[sched_ctx].total_elapsed_flops[workerid] += diff_elapsed_flops;
  1174. // starpu_pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
  1175. }
  1176. if(starpu_sched_ctx_get_hierarchy_level(sched_ctx) > 0)
  1177. sc_hypervisor_update_diff_elapsed_flops(starpu_sched_ctx_get_inheritor(sched_ctx), diff_elapsed_flops);
  1178. }
  1179. void sc_hypervisor_get_ctxs_on_level(unsigned **sched_ctxs, int *nsched_ctxs, unsigned hierarchy_level, unsigned father_sched_ctx_id)
  1180. {
  1181. unsigned s;
  1182. *nsched_ctxs = 0;
  1183. *sched_ctxs = (unsigned*)malloc(hypervisor.nsched_ctxs * sizeof(unsigned));
  1184. for(s = 0; s < hypervisor.nsched_ctxs; s++)
  1185. {
  1186. /* if father == STARPU_NMAX_SCHED_CTXS we take all the ctxs in this level */
  1187. if(starpu_sched_ctx_get_hierarchy_level(hypervisor.sched_ctxs[s]) == hierarchy_level &&
  1188. (starpu_sched_ctx_get_inheritor(hypervisor.sched_ctxs[s]) == father_sched_ctx_id || father_sched_ctx_id == STARPU_NMAX_SCHED_CTXS))
  1189. (*sched_ctxs)[(*nsched_ctxs)++] = hypervisor.sched_ctxs[s];
  1190. }
  1191. if(*nsched_ctxs == 0)
  1192. free(*sched_ctxs);
  1193. return;
  1194. }
  1195. unsigned sc_hypervisor_get_nhierarchy_levels(void)
  1196. {
  1197. unsigned nlevels = 0;
  1198. unsigned level = 0;
  1199. unsigned levels[STARPU_NMAX_SCHED_CTXS];
  1200. unsigned s, l;
  1201. for(s = 0; s < hypervisor.nsched_ctxs; s++)
  1202. {
  1203. level = starpu_sched_ctx_get_hierarchy_level(hypervisor.sched_ctxs[s]);
  1204. unsigned found = 0;
  1205. for(l = 0; l < nlevels; l++)
  1206. if(levels[l] == level)
  1207. found = 1;
  1208. if(!found)
  1209. levels[nlevels++] = level;
  1210. }
  1211. return nlevels;
  1212. }