sc_hypervisor.c 42 KB

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