sc_hypervisor.c 57 KB

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