sc_hypervisor.c 57 KB

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