sched_ctx.c 81 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2011, 2013 INRIA
  4. * Copyright (C) 2016 Uppsala University
  5. *
  6. * StarPU is free software; you can redistribute it and/or modify
  7. * it under the terms of the GNU Lesser General Public License as published by
  8. * the Free Software Foundation; either version 2.1 of the License, or (at
  9. * your option) any later version.
  10. *
  11. * StarPU is distributed in the hope that it will be useful, but
  12. * WITHOUT ANY WARRANTY; without even the implied warranty of
  13. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  14. *
  15. * See the GNU Lesser General Public License in COPYING.LGPL for more details.
  16. */
  17. #include <core/sched_policy.h>
  18. #include <core/sched_ctx.h>
  19. #include <common/utils.h>
  20. #include <stdarg.h>
  21. starpu_pthread_rwlock_t changing_ctx_mutex[STARPU_NMAX_SCHED_CTXS];
  22. static starpu_pthread_mutex_t sched_ctx_manag = STARPU_PTHREAD_MUTEX_INITIALIZER;
  23. static starpu_pthread_mutex_t finished_submit_mutex = STARPU_PTHREAD_MUTEX_INITIALIZER;
  24. static struct starpu_task stop_submission_task = STARPU_TASK_INITIALIZER;
  25. starpu_pthread_key_t sched_ctx_key;
  26. static unsigned with_hypervisor = 0;
  27. static double hyp_start_sample[STARPU_NMAX_SCHED_CTXS];
  28. static double hyp_start_allow_sample[STARPU_NMAX_SCHED_CTXS];
  29. static double flops[STARPU_NMAX_SCHED_CTXS][STARPU_NMAXWORKERS];
  30. static size_t data_size[STARPU_NMAX_SCHED_CTXS][STARPU_NMAXWORKERS];
  31. static double hyp_actual_start_sample[STARPU_NMAX_SCHED_CTXS];
  32. static double window_size;
  33. static int nobind;
  34. static int occupied_sms = 0;
  35. static unsigned _starpu_get_first_free_sched_ctx(struct _starpu_machine_config *config);
  36. static void _starpu_sched_ctx_put_new_master(unsigned sched_ctx_id);
  37. static void _starpu_sched_ctx_put_workers_to_sleep(unsigned sched_ctx_id, unsigned all);
  38. static void _starpu_sched_ctx_wake_up_workers(unsigned sched_ctx_id, unsigned all);
  39. static void _starpu_sched_ctx_update_parallel_workers_with(unsigned sched_ctx_id);
  40. static void _starpu_sched_ctx_update_parallel_workers_without(unsigned sched_ctx_id);
  41. static void _starpu_worker_gets_into_ctx(unsigned sched_ctx_id, struct _starpu_worker *worker)
  42. {
  43. unsigned ret_sched_ctx = _starpu_sched_ctx_elt_exists(worker->sched_ctx_list, sched_ctx_id);
  44. /* the worker was planning to go away in another ctx but finally he changed his mind &
  45. he's staying */
  46. if (!ret_sched_ctx)
  47. {
  48. /* add context to worker */
  49. _starpu_sched_ctx_list_add(&worker->sched_ctx_list, sched_ctx_id);
  50. worker->nsched_ctxs++;
  51. }
  52. worker->removed_from_ctx[sched_ctx_id] = 0;
  53. if(worker->tmp_sched_ctx == (int) sched_ctx_id)
  54. worker->tmp_sched_ctx = -1;
  55. return;
  56. }
  57. void _starpu_worker_gets_out_of_ctx(unsigned sched_ctx_id, struct _starpu_worker *worker)
  58. {
  59. unsigned ret_sched_ctx = _starpu_sched_ctx_elt_exists(worker->sched_ctx_list, sched_ctx_id);
  60. /* remove context from worker */
  61. if(ret_sched_ctx)
  62. {
  63. /* don't remove scheduling data here, there might be tasks running and when post_exec
  64. executes scheduling data is not there any more, do it when deleting context, then
  65. we really won't need it anymore */
  66. /* struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id); */
  67. /* if(sched_ctx && sched_ctx->sched_policy && sched_ctx->sched_policy->remove_workers) */
  68. /* { */
  69. /* _STARPU_TRACE_WORKER_SCHEDULING_PUSH; */
  70. /* sched_ctx->sched_policy->remove_workers(sched_ctx_id, &worker->workerid, 1); */
  71. /* _STARPU_TRACE_WORKER_SCHEDULING_POP; */
  72. /* } */
  73. if (!_starpu_sched_ctx_list_remove(&worker->sched_ctx_list, sched_ctx_id))
  74. worker->nsched_ctxs--;
  75. }
  76. return;
  77. }
  78. static void _starpu_update_workers_with_ctx(int *workerids, int nworkers, int sched_ctx_id)
  79. {
  80. int i;
  81. struct _starpu_worker *worker = NULL;
  82. for(i = 0; i < nworkers; i++)
  83. {
  84. worker = _starpu_get_worker_struct(workerids[i]);
  85. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
  86. _starpu_worker_gets_into_ctx(sched_ctx_id, worker);
  87. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
  88. }
  89. return;
  90. }
  91. static void _starpu_update_workers_without_ctx(int *workerids, int nworkers, int sched_ctx_id, unsigned now)
  92. {
  93. int i;
  94. struct _starpu_worker *worker = NULL;
  95. for(i = 0; i < nworkers; i++)
  96. {
  97. worker = _starpu_get_worker_struct(workerids[i]);
  98. if(now)
  99. {
  100. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
  101. _starpu_worker_gets_out_of_ctx(sched_ctx_id, worker);
  102. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
  103. }
  104. else
  105. {
  106. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
  107. worker->removed_from_ctx[sched_ctx_id] = 1;
  108. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
  109. }
  110. }
  111. return;
  112. }
  113. void starpu_sched_ctx_stop_task_submission()
  114. {
  115. _starpu_exclude_task_from_dag(&stop_submission_task);
  116. int ret = _starpu_task_submit_internally(&stop_submission_task);
  117. STARPU_ASSERT(!ret);
  118. }
  119. void starpu_sched_ctx_worker_shares_tasks_lists(int workerid, int sched_ctx_id)
  120. {
  121. struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
  122. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  123. int curr_workerid = starpu_worker_get_id();
  124. /* if is the initial sched_ctx no point in taking the mutex, the workers are
  125. not launched yet, or if the current worker is calling this */
  126. if(!sched_ctx->is_initial_sched && workerid != curr_workerid)
  127. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
  128. worker->shares_tasks_lists[sched_ctx_id] = 1;
  129. if(!sched_ctx->is_initial_sched && workerid != curr_workerid)
  130. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
  131. }
  132. static void _starpu_add_workers_to_sched_ctx(struct _starpu_sched_ctx *sched_ctx, int *workerids, int nworkers,
  133. int *added_workers, int *n_added_workers)
  134. {
  135. struct starpu_worker_collection *workers = sched_ctx->workers;
  136. struct _starpu_machine_config *config = _starpu_get_machine_config();
  137. int nworkers_to_add = nworkers == -1 ? (int)config->topology.nworkers : nworkers;
  138. if (!nworkers_to_add)
  139. return;
  140. int workers_to_add[nworkers_to_add];
  141. struct starpu_perfmodel_device devices[nworkers_to_add];
  142. int ndevices = 0;
  143. struct _starpu_worker *str_worker = NULL;
  144. int worker;
  145. int i = 0;
  146. for(i = 0; i < nworkers_to_add; i++)
  147. {
  148. /* added_workers is NULL for the call of this func at the creation of the context*/
  149. /* if the function is called at the creation of the context it's no need to do this verif */
  150. if(added_workers)
  151. {
  152. worker = workers->add(workers, (workerids == NULL ? i : workerids[i]));
  153. if(worker >= 0)
  154. added_workers[(*n_added_workers)++] = worker;
  155. else
  156. {
  157. int curr_workerid = starpu_worker_get_id();
  158. struct _starpu_worker *worker_str = _starpu_get_worker_struct(workerids[i]);
  159. if(curr_workerid != workerids[i])
  160. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker_str->sched_mutex);
  161. worker_str->removed_from_ctx[sched_ctx->id] = 0;
  162. if(curr_workerid != workerids[i])
  163. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker_str->sched_mutex);
  164. }
  165. }
  166. else
  167. {
  168. worker = (workerids == NULL ? i : workerids[i]);
  169. workers->add(workers, worker);
  170. workers_to_add[i] = worker;
  171. str_worker = _starpu_get_worker_struct(worker);
  172. str_worker->tmp_sched_ctx = (int)sched_ctx->id;
  173. }
  174. }
  175. int *wa;
  176. int na;
  177. if(added_workers)
  178. {
  179. na = *n_added_workers;
  180. wa = added_workers;
  181. }
  182. else
  183. {
  184. na = nworkers_to_add;
  185. wa = workers_to_add;
  186. }
  187. for(i = 0; i < na; i++)
  188. {
  189. worker = wa[i];
  190. str_worker = _starpu_get_worker_struct(worker);
  191. int dev1, dev2;
  192. unsigned found = 0;
  193. for(dev1 = 0; dev1 < str_worker->perf_arch.ndevices; dev1++)
  194. {
  195. for(dev2 = 0; dev2 < ndevices; dev2++)
  196. {
  197. if(devices[dev2].type == str_worker->perf_arch.devices[dev1].type &&
  198. devices[dev2].devid == str_worker->perf_arch.devices[dev1].devid)
  199. {
  200. devices[dev2].ncores += str_worker->perf_arch.devices[dev1].ncores;
  201. found = 1;
  202. break;
  203. }
  204. }
  205. if(!found)
  206. {
  207. devices[ndevices].type = str_worker->perf_arch.devices[dev1].type;
  208. devices[ndevices].devid = str_worker->perf_arch.devices[dev1].devid;
  209. devices[ndevices].ncores = str_worker->perf_arch.devices[dev1].ncores;
  210. ndevices++;
  211. }
  212. else
  213. found = 0;
  214. }
  215. }
  216. if(ndevices > 0)
  217. {
  218. if(sched_ctx->perf_arch.devices == NULL)
  219. {
  220. _STARPU_MALLOC(sched_ctx->perf_arch.devices, ndevices*sizeof(struct starpu_perfmodel_device));
  221. }
  222. else
  223. {
  224. int nfinal_devices = 0;
  225. int dev1, dev2;
  226. unsigned found = 0;
  227. for(dev1 = 0; dev1 < ndevices; dev1++)
  228. {
  229. for(dev2 = 0; dev2 < sched_ctx->perf_arch.ndevices; dev2++)
  230. {
  231. if(sched_ctx->perf_arch.devices[dev2].type == devices[dev1].type && sched_ctx->perf_arch.devices[dev2].devid == devices[dev1].devid)
  232. found = 1;
  233. }
  234. if(!found)
  235. {
  236. nfinal_devices++;
  237. }
  238. else
  239. found = 0;
  240. }
  241. int nsize = (sched_ctx->perf_arch.ndevices+nfinal_devices);
  242. _STARPU_REALLOC(sched_ctx->perf_arch.devices, nsize*sizeof(struct starpu_perfmodel_device));
  243. }
  244. int dev1, dev2;
  245. unsigned found = 0;
  246. for(dev1 = 0; dev1 < ndevices; dev1++)
  247. {
  248. for(dev2 = 0; dev2 < sched_ctx->perf_arch.ndevices; dev2++)
  249. {
  250. if(sched_ctx->perf_arch.devices[dev2].type == devices[dev1].type && sched_ctx->perf_arch.devices[dev2].devid == devices[dev1].devid)
  251. {
  252. if(sched_ctx->perf_arch.devices[dev2].type == STARPU_CPU_WORKER)
  253. sched_ctx->perf_arch.devices[dev2].ncores += devices[dev1].ncores;
  254. found = 1;
  255. }
  256. }
  257. if(!found)
  258. {
  259. sched_ctx->perf_arch.devices[sched_ctx->perf_arch.ndevices].type = devices[dev1].type;
  260. sched_ctx->perf_arch.devices[sched_ctx->perf_arch.ndevices].devid = devices[dev1].devid;
  261. if (sched_ctx->stream_worker != -1)
  262. sched_ctx->perf_arch.devices[sched_ctx->perf_arch.ndevices].ncores = sched_ctx->nsms;
  263. else
  264. sched_ctx->perf_arch.devices[sched_ctx->perf_arch.ndevices].ncores = devices[dev1].ncores;
  265. sched_ctx->perf_arch.ndevices++;
  266. }
  267. else
  268. found = 0;
  269. }
  270. }
  271. _starpu_sched_ctx_update_parallel_workers_with(sched_ctx->id);
  272. if(sched_ctx->sched_policy && sched_ctx->sched_policy->add_workers)
  273. {
  274. _STARPU_TRACE_WORKER_SCHEDULING_PUSH;
  275. if(added_workers)
  276. {
  277. if(*n_added_workers > 0)
  278. sched_ctx->sched_policy->add_workers(sched_ctx->id, added_workers, *n_added_workers);
  279. }
  280. else
  281. {
  282. sched_ctx->sched_policy->add_workers(sched_ctx->id, workers_to_add, nworkers_to_add);
  283. }
  284. _STARPU_TRACE_WORKER_SCHEDULING_POP;
  285. }
  286. return;
  287. }
  288. static void _starpu_remove_workers_from_sched_ctx(struct _starpu_sched_ctx *sched_ctx, int *workerids,
  289. int nworkers, int *removed_workers, int *n_removed_workers)
  290. {
  291. struct starpu_worker_collection *workers = sched_ctx->workers;
  292. struct starpu_perfmodel_device devices[workers->nworkers];
  293. int ndevices = 0;
  294. int i = 0;
  295. for(i = 0; i < nworkers; i++)
  296. {
  297. if(workers->nworkers > 0)
  298. {
  299. if(_starpu_worker_belongs_to_a_sched_ctx(workerids[i], sched_ctx->id))
  300. {
  301. int worker = workers->remove(workers, workerids[i]);
  302. if(worker >= 0)
  303. removed_workers[(*n_removed_workers)++] = worker;
  304. }
  305. }
  306. }
  307. unsigned found = 0;
  308. int dev;
  309. struct starpu_sched_ctx_iterator it;
  310. if(workers->init_iterator)
  311. workers->init_iterator(workers, &it);
  312. while(workers->has_next(workers, &it))
  313. {
  314. int worker = workers->get_next(workers, &it);
  315. struct _starpu_worker *str_worker = _starpu_get_worker_struct(worker);
  316. for(dev = 0; dev < str_worker->perf_arch.ndevices; dev++)
  317. {
  318. int dev2;
  319. for(dev2 = 0; dev2 < ndevices; dev2++)
  320. {
  321. if(devices[dev2].type == str_worker->perf_arch.devices[dev].type &&
  322. devices[dev2].devid == str_worker->perf_arch.devices[dev].devid)
  323. {
  324. if(devices[dev2].type == STARPU_CPU_WORKER)
  325. devices[dev2].ncores += str_worker->perf_arch.devices[dev].ncores;
  326. }
  327. found = 1;
  328. }
  329. if(!found)
  330. {
  331. devices[ndevices].type = str_worker->perf_arch.devices[dev].type;
  332. devices[ndevices].devid = str_worker->perf_arch.devices[dev].devid;
  333. devices[ndevices].ncores = str_worker->perf_arch.devices[dev].ncores;
  334. ndevices++;
  335. }
  336. else
  337. found = 0;
  338. }
  339. found = 0;
  340. }
  341. sched_ctx->perf_arch.ndevices = ndevices;
  342. for(dev = 0; dev < ndevices; dev++)
  343. {
  344. sched_ctx->perf_arch.devices[dev].type = devices[dev].type;
  345. sched_ctx->perf_arch.devices[dev].devid = devices[dev].devid;
  346. sched_ctx->perf_arch.devices[dev].ncores = devices[dev].ncores;
  347. }
  348. _starpu_sched_ctx_update_parallel_workers_without(sched_ctx->id);
  349. return;
  350. }
  351. static void _starpu_sched_ctx_free_scheduling_data(struct _starpu_sched_ctx *sched_ctx)
  352. {
  353. if(sched_ctx->sched_policy && sched_ctx->sched_policy->remove_workers)
  354. {
  355. int *workerids = NULL;
  356. unsigned nworkers_ctx = starpu_sched_ctx_get_workers_list(sched_ctx->id, &workerids);
  357. if(nworkers_ctx > 0)
  358. {
  359. _STARPU_TRACE_WORKER_SCHEDULING_PUSH;
  360. sched_ctx->sched_policy->remove_workers(sched_ctx->id, workerids, nworkers_ctx);
  361. _STARPU_TRACE_WORKER_SCHEDULING_POP;
  362. }
  363. free(workerids);
  364. }
  365. return;
  366. }
  367. #ifdef STARPU_HAVE_HWLOC
  368. static void _starpu_sched_ctx_create_hwloc_tree(struct _starpu_sched_ctx *sched_ctx)
  369. {
  370. sched_ctx->hwloc_workers_set = hwloc_bitmap_alloc();
  371. struct starpu_worker_collection *workers = sched_ctx->workers;
  372. struct _starpu_worker *worker;
  373. struct starpu_sched_ctx_iterator it;
  374. workers->init_iterator(workers, &it);
  375. while(workers->has_next(workers, &it))
  376. {
  377. unsigned workerid = workers->get_next(workers, &it);
  378. if(!starpu_worker_is_combined_worker(workerid))
  379. {
  380. worker = _starpu_get_worker_struct(workerid);
  381. hwloc_bitmap_or(sched_ctx->hwloc_workers_set,
  382. sched_ctx->hwloc_workers_set,
  383. worker->hwloc_cpu_set);
  384. }
  385. }
  386. return;
  387. }
  388. #endif
  389. struct _starpu_sched_ctx* _starpu_create_sched_ctx(struct starpu_sched_policy *policy, int *workerids,
  390. int nworkers_ctx, unsigned is_initial_sched,
  391. const char *sched_ctx_name,
  392. int min_prio_set, int min_prio,
  393. int max_prio_set, int max_prio,
  394. unsigned awake_workers,
  395. void (*sched_policy_init)(unsigned),
  396. void * user_data,
  397. int nsub_ctxs, int *sub_ctxs, int nsms)
  398. {
  399. struct _starpu_machine_config *config = _starpu_get_machine_config();
  400. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx_manag);
  401. STARPU_ASSERT(config->topology.nsched_ctxs < STARPU_NMAX_SCHED_CTXS);
  402. unsigned id = _starpu_get_first_free_sched_ctx(config);
  403. struct _starpu_sched_ctx *sched_ctx = &config->sched_ctxs[id];
  404. sched_ctx->id = id;
  405. config->topology.nsched_ctxs++;
  406. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx_manag);
  407. int nworkers = config->topology.nworkers;
  408. STARPU_ASSERT(nworkers_ctx <= nworkers);
  409. STARPU_PTHREAD_MUTEX_INIT(&sched_ctx->empty_ctx_mutex, NULL);
  410. starpu_task_list_init(&sched_ctx->empty_ctx_tasks);
  411. STARPU_PTHREAD_MUTEX_INIT(&sched_ctx->waiting_tasks_mutex, NULL);
  412. starpu_task_list_init(&sched_ctx->waiting_tasks);
  413. STARPU_PTHREAD_MUTEX_INIT(&sched_ctx->sched_ctx_list_mutex, NULL);
  414. if (policy)
  415. {
  416. _STARPU_MALLOC(sched_ctx->sched_policy, sizeof(struct starpu_sched_policy));
  417. }
  418. else
  419. {
  420. sched_ctx->sched_policy = NULL;
  421. }
  422. sched_ctx->is_initial_sched = is_initial_sched;
  423. sched_ctx->name = sched_ctx_name;
  424. sched_ctx->inheritor = STARPU_NMAX_SCHED_CTXS;
  425. sched_ctx->finished_submit = 0;
  426. sched_ctx->min_priority_is_set = min_prio_set;
  427. if (sched_ctx->min_priority_is_set) sched_ctx->min_priority = min_prio;
  428. sched_ctx->max_priority_is_set = max_prio_set;
  429. if (sched_ctx->max_priority_is_set) sched_ctx->max_priority = max_prio;
  430. _starpu_barrier_counter_init(&sched_ctx->tasks_barrier, 0);
  431. _starpu_barrier_counter_init(&sched_ctx->ready_tasks_barrier, 0);
  432. sched_ctx->ready_flops = 0.0;
  433. sched_ctx->main_master = -1;
  434. sched_ctx->perf_arch.devices = NULL;
  435. sched_ctx->perf_arch.ndevices = 0;
  436. sched_ctx->init_sched = sched_policy_init;
  437. sched_ctx->user_data = user_data;
  438. sched_ctx->sms_start_idx = 0;
  439. sched_ctx->sms_end_idx = STARPU_NMAXSMS;
  440. sched_ctx->nsms = nsms;
  441. sched_ctx->stream_worker = -1;
  442. if(nsms > 0)
  443. {
  444. STARPU_ASSERT_MSG(workerids, "workerids is needed when setting nsms");
  445. sched_ctx->sms_start_idx = occupied_sms;
  446. sched_ctx->sms_end_idx = occupied_sms+nsms;
  447. occupied_sms += nsms;
  448. _STARPU_DEBUG("ctx %d: stream worker %d nsms %d ocupied sms %d\n", sched_ctx->id, workerids[0], nsms, occupied_sms);
  449. STARPU_ASSERT_MSG(occupied_sms <= STARPU_NMAXSMS , "STARPU:requested more sms than available");
  450. _starpu_worker_set_stream_ctx(workerids[0], sched_ctx);
  451. sched_ctx->stream_worker = workerids[0];
  452. }
  453. sched_ctx->nsub_ctxs = 0;
  454. int w;
  455. for(w = 0; w < nworkers; w++)
  456. {
  457. sem_init(&sched_ctx->fall_asleep_sem[w], 0, 0);
  458. sem_init(&sched_ctx->wake_up_sem[w], 0, 0);
  459. STARPU_PTHREAD_COND_INIT(&sched_ctx->parallel_sect_cond[w], NULL);
  460. STARPU_PTHREAD_MUTEX_INIT(&sched_ctx->parallel_sect_mutex[w], NULL);
  461. STARPU_PTHREAD_COND_INIT(&sched_ctx->parallel_sect_cond_busy[w], NULL);
  462. sched_ctx->busy[w] = 0;
  463. sched_ctx->parallel_sect[w] = 0;
  464. sched_ctx->sleeping[w] = 0;
  465. }
  466. sched_ctx->parallel_view = 0;
  467. /*init the strategy structs and the worker_collection of the ressources of the context */
  468. if(policy)
  469. {
  470. _starpu_init_sched_policy(config, sched_ctx, policy);
  471. sched_ctx->awake_workers = 1;
  472. }
  473. else
  474. {
  475. sched_ctx->awake_workers = awake_workers;
  476. starpu_sched_ctx_create_worker_collection(sched_ctx->id, STARPU_WORKER_LIST);
  477. }
  478. if(is_initial_sched)
  479. {
  480. int i;
  481. /*initialize the mutexes for all contexts */
  482. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  483. {
  484. STARPU_PTHREAD_RWLOCK_INIT(&changing_ctx_mutex[i], NULL);
  485. }
  486. }
  487. /*add sub_ctxs before add workers, in order to be able to associate them if necessary */
  488. if(nsub_ctxs != 0)
  489. {
  490. int i;
  491. for(i = 0; i < nsub_ctxs; i++)
  492. sched_ctx->sub_ctxs[i] = sub_ctxs[i];
  493. sched_ctx->nsub_ctxs = nsub_ctxs;
  494. }
  495. /* after having an worker_collection on the ressources add them */
  496. _starpu_add_workers_to_sched_ctx(sched_ctx, workerids, nworkers_ctx, NULL, NULL);
  497. #ifdef STARPU_HAVE_HWLOC
  498. /* build hwloc tree of the context */
  499. _starpu_sched_ctx_create_hwloc_tree(sched_ctx);
  500. #endif //STARPU_HAVE_HWLOC
  501. /* if we create the initial big sched ctx we can update workers' status here
  502. because they haven't been launched yet */
  503. if(is_initial_sched)
  504. {
  505. int i;
  506. for(i = 0; i < nworkers; i++)
  507. {
  508. struct _starpu_worker *worker = _starpu_get_worker_struct(i);
  509. if(!_starpu_sched_ctx_list_add(&worker->sched_ctx_list, sched_ctx->id))
  510. worker->nsched_ctxs++;
  511. }
  512. }
  513. return sched_ctx;
  514. }
  515. static void _get_workers(int min, int max, int *workers, int *nw, enum starpu_worker_archtype arch, unsigned allow_overlap)
  516. {
  517. int pus[max];
  518. int npus = 0;
  519. int i;
  520. struct _starpu_machine_config *config = _starpu_get_machine_config();
  521. if(config->topology.nsched_ctxs == 1)
  522. {
  523. /*we have all available resources */
  524. npus = starpu_worker_get_nids_by_type(arch, pus, max);
  525. /*TODO: hierarchical ctxs: get max good workers: close one to another */
  526. for(i = 0; i < npus; i++)
  527. workers[(*nw)++] = pus[i];
  528. }
  529. else
  530. {
  531. unsigned enough_ressources = 0;
  532. npus = starpu_worker_get_nids_ctx_free_by_type(arch, pus, max);
  533. for(i = 0; i < npus; i++)
  534. workers[(*nw)++] = pus[i];
  535. if(npus == max)
  536. /*we have enough available resources */
  537. enough_ressources = 1;
  538. if(!enough_ressources && npus >= min)
  539. /*we have enough available resources */
  540. enough_ressources = 1;
  541. if(!enough_ressources)
  542. {
  543. /* try to get ressources from ctx who have more than the min of workers they need */
  544. int s;
  545. for(s = 1; s < STARPU_NMAX_SCHED_CTXS; s++)
  546. {
  547. if(config->sched_ctxs[s].id != STARPU_NMAX_SCHED_CTXS)
  548. {
  549. int _npus = 0;
  550. int _pus[STARPU_NMAXWORKERS];
  551. _npus = _starpu_get_workers_of_sched_ctx(config->sched_ctxs[s].id, _pus, arch);
  552. int ctx_min = arch == STARPU_CPU_WORKER ? config->sched_ctxs[s].min_ncpus : config->sched_ctxs[s].min_ngpus;
  553. if(_npus > ctx_min)
  554. {
  555. int n=0;
  556. if(npus < min)
  557. {
  558. n = (_npus - ctx_min) > (min - npus) ? min - npus : (_npus - ctx_min);
  559. npus += n;
  560. }
  561. /*TODO: hierarchical ctxs: get n good workers: close to the other ones I already assigned to the ctx */
  562. for(i = 0; i < n; i++)
  563. workers[(*nw)++] = _pus[i];
  564. starpu_sched_ctx_remove_workers(_pus, n, config->sched_ctxs[s].id);
  565. }
  566. }
  567. }
  568. if(npus >= min)
  569. enough_ressources = 1;
  570. }
  571. if(!enough_ressources)
  572. {
  573. /* if there is no available workers to satisfy the minimum required
  574. give them workers proportional to their requirements*/
  575. int global_npus = starpu_worker_get_count_by_type(arch);
  576. int req_npus = 0;
  577. int s;
  578. for(s = 1; s < STARPU_NMAX_SCHED_CTXS; s++)
  579. if(config->sched_ctxs[s].id != STARPU_NMAX_SCHED_CTXS)
  580. req_npus += arch == STARPU_CPU_WORKER ? config->sched_ctxs[s].min_ncpus : config->sched_ctxs[s].min_ngpus;
  581. req_npus += min;
  582. for(s = 1; s < STARPU_NMAX_SCHED_CTXS; s++)
  583. {
  584. if(config->sched_ctxs[s].id != STARPU_NMAX_SCHED_CTXS)
  585. {
  586. int ctx_min = arch == STARPU_CPU_WORKER ? config->sched_ctxs[s].min_ncpus : config->sched_ctxs[s].min_ngpus;
  587. double needed_npus = ((double)ctx_min * (double)global_npus) / (double)req_npus;
  588. int _npus = 0;
  589. int _pus[STARPU_NMAXWORKERS];
  590. _npus = _starpu_get_workers_of_sched_ctx(config->sched_ctxs[s].id, _pus, arch);
  591. if(needed_npus < (double)_npus)
  592. {
  593. double npus_to_rem = (double)_npus - needed_npus;
  594. int x = floor(npus_to_rem);
  595. double x_double = (double)x;
  596. double diff = npus_to_rem - x_double;
  597. int npus_to_remove = diff >= 0.5 ? x+1 : x;
  598. int pus_to_remove[npus_to_remove];
  599. int c = 0;
  600. /*TODO: hierarchical ctxs: get npus_to_remove good workers: close to the other ones I already assigned to the ctx */
  601. for(i = _npus-1; i >= (_npus - npus_to_remove); i--)
  602. {
  603. workers[(*nw)++] = _pus[i];
  604. pus_to_remove[c++] = _pus[i];
  605. }
  606. if(!allow_overlap)
  607. starpu_sched_ctx_remove_workers(pus_to_remove, npus_to_remove, config->sched_ctxs[s].id);
  608. }
  609. }
  610. }
  611. }
  612. }
  613. }
  614. unsigned starpu_sched_ctx_create_inside_interval(const char *policy_name, const char *sched_ctx_name,
  615. int min_ncpus, int max_ncpus, int min_ngpus, int max_ngpus,
  616. unsigned allow_overlap)
  617. {
  618. struct _starpu_machine_config *config = _starpu_get_machine_config();
  619. struct starpu_sched_policy *selected_policy = _starpu_select_sched_policy(config, policy_name);
  620. struct _starpu_sched_ctx *sched_ctx = NULL;
  621. int workers[max_ncpus + max_ngpus];
  622. int nw = 0;
  623. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx_manag);
  624. _get_workers(min_ncpus, max_ncpus, workers, &nw, STARPU_CPU_WORKER, allow_overlap);
  625. _get_workers(min_ngpus, max_ngpus, workers, &nw, STARPU_CUDA_WORKER, allow_overlap);
  626. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx_manag);
  627. int i;
  628. printf("%d: ", nw);
  629. for(i = 0; i < nw; i++)
  630. printf("%d ", workers[i]);
  631. printf("\n");
  632. sched_ctx = _starpu_create_sched_ctx(selected_policy, workers, nw, 0, sched_ctx_name, 0, 0, 0, 0, 1, NULL, NULL,0, NULL, 0);
  633. sched_ctx->min_ncpus = min_ncpus;
  634. sched_ctx->max_ncpus = max_ncpus;
  635. sched_ctx->min_ngpus = min_ngpus;
  636. sched_ctx->max_ngpus = max_ngpus;
  637. _starpu_unlock_mutex_if_prev_locked();
  638. int *added_workerids;
  639. unsigned nw_ctx = starpu_sched_ctx_get_workers_list(sched_ctx->id, &added_workerids);
  640. _starpu_update_workers_without_ctx(added_workerids, nw_ctx, sched_ctx->id, 0);
  641. free(added_workerids);
  642. _starpu_relock_mutex_if_prev_locked();
  643. #ifdef STARPU_USE_SC_HYPERVISOR
  644. sched_ctx->perf_counters = NULL;
  645. #endif
  646. return sched_ctx->id;
  647. }
  648. int starpu_sched_ctx_get_nsms(unsigned sched_ctx)
  649. {
  650. struct _starpu_sched_ctx *sc = _starpu_get_sched_ctx_struct(sched_ctx);
  651. return sc->nsms;
  652. }
  653. void starpu_sched_ctx_get_sms_interval(int stream_workerid, int *start, int *end)
  654. {
  655. struct _starpu_sched_ctx *sc = _starpu_worker_get_ctx_stream(stream_workerid);
  656. *start = sc->sms_start_idx;
  657. *end = sc->sms_end_idx;
  658. }
  659. int starpu_sched_ctx_get_sub_ctxs(unsigned sched_ctx, int *ctxs)
  660. {
  661. struct _starpu_sched_ctx *sc = _starpu_get_sched_ctx_struct(sched_ctx);
  662. int i;
  663. for(i = 0; i < sc->nsub_ctxs; i++)
  664. ctxs[i] = sc->sub_ctxs[i];
  665. return sc->nsub_ctxs;
  666. }
  667. int starpu_sched_ctx_get_stream_worker(unsigned sub_ctx)
  668. {
  669. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sub_ctx);
  670. struct starpu_worker_collection *workers = sched_ctx->workers;
  671. struct starpu_sched_ctx_iterator it;
  672. int worker = -1;
  673. workers->init_iterator(workers, &it);
  674. if(workers->has_next(workers, &it))
  675. {
  676. worker = workers->get_next(workers, &it);
  677. }
  678. return worker;
  679. }
  680. unsigned starpu_sched_ctx_create(int *workerids, int nworkers, const char *sched_ctx_name, ...)
  681. {
  682. va_list varg_list;
  683. int arg_type;
  684. int min_prio_set = 0;
  685. int max_prio_set = 0;
  686. int min_prio = 0;
  687. int max_prio = 0;
  688. int nsms = 0;
  689. int *sub_ctxs = NULL;
  690. int nsub_ctxs = 0;
  691. void *user_data = NULL;
  692. struct starpu_sched_policy *sched_policy = NULL;
  693. unsigned hierarchy_level = 0;
  694. unsigned nesting_sched_ctx = STARPU_NMAX_SCHED_CTXS;
  695. unsigned awake_workers = 0;
  696. void (*init_sched)(unsigned) = NULL;
  697. va_start(varg_list, sched_ctx_name);
  698. while ((arg_type = va_arg(varg_list, int)) != 0)
  699. {
  700. if (arg_type == STARPU_SCHED_CTX_POLICY_NAME)
  701. {
  702. char *policy_name = va_arg(varg_list, char *);
  703. struct _starpu_machine_config *config = _starpu_get_machine_config();
  704. sched_policy = _starpu_select_sched_policy(config, policy_name);
  705. }
  706. else if (arg_type == STARPU_SCHED_CTX_POLICY_STRUCT)
  707. {
  708. sched_policy = va_arg(varg_list, struct starpu_sched_policy *);
  709. }
  710. else if (arg_type == STARPU_SCHED_CTX_POLICY_MIN_PRIO)
  711. {
  712. min_prio = va_arg(varg_list, int);
  713. min_prio_set = 1;
  714. }
  715. else if (arg_type == STARPU_SCHED_CTX_POLICY_MAX_PRIO)
  716. {
  717. max_prio = va_arg(varg_list, int);
  718. max_prio_set = 1;
  719. }
  720. else if (arg_type == STARPU_SCHED_CTX_HIERARCHY_LEVEL)
  721. {
  722. hierarchy_level = va_arg(varg_list, unsigned);
  723. }
  724. else if (arg_type == STARPU_SCHED_CTX_NESTED)
  725. {
  726. nesting_sched_ctx = va_arg(varg_list, unsigned);
  727. }
  728. else if (arg_type == STARPU_SCHED_CTX_AWAKE_WORKERS)
  729. {
  730. awake_workers = 1;
  731. }
  732. else if (arg_type == STARPU_SCHED_CTX_POLICY_INIT)
  733. {
  734. init_sched = va_arg(varg_list, void(*)(unsigned));
  735. }
  736. else if (arg_type == STARPU_SCHED_CTX_USER_DATA)
  737. {
  738. user_data = va_arg(varg_list, void *);
  739. }
  740. else if (arg_type == STARPU_SCHED_CTX_SUB_CTXS)
  741. {
  742. sub_ctxs = va_arg(varg_list, int*);
  743. nsub_ctxs = va_arg(varg_list, int);
  744. }
  745. else if (arg_type == STARPU_SCHED_CTX_CUDA_NSMS)
  746. {
  747. nsms = va_arg(varg_list, int);
  748. }
  749. else
  750. {
  751. STARPU_ABORT_MSG("Unrecognized argument %d\n", arg_type);
  752. }
  753. }
  754. va_end(varg_list);
  755. if (workerids && nworkers != -1)
  756. {
  757. /* Make sure the user doesn't use invalid worker IDs. */
  758. int num_workers = starpu_worker_get_count();
  759. int i;
  760. for (i = 0; i < nworkers; i++)
  761. {
  762. if (workerids[i] < 0 || workerids[i] >= num_workers)
  763. {
  764. _STARPU_ERROR("Invalid worker ID (%d) specified!\n", workerids[i]);
  765. return STARPU_NMAX_SCHED_CTXS;
  766. }
  767. }
  768. }
  769. struct _starpu_sched_ctx *sched_ctx = NULL;
  770. sched_ctx = _starpu_create_sched_ctx(sched_policy, workerids, nworkers, 0, sched_ctx_name, min_prio_set, min_prio, max_prio_set, max_prio, awake_workers, init_sched, user_data, nsub_ctxs, sub_ctxs, nsms);
  771. sched_ctx->hierarchy_level = hierarchy_level;
  772. sched_ctx->nesting_sched_ctx = nesting_sched_ctx;
  773. _starpu_unlock_mutex_if_prev_locked();
  774. int *added_workerids;
  775. unsigned nw_ctx = starpu_sched_ctx_get_workers_list(sched_ctx->id, &added_workerids);
  776. _starpu_update_workers_with_ctx(added_workerids, nw_ctx, sched_ctx->id);
  777. free(added_workerids);
  778. _starpu_relock_mutex_if_prev_locked();
  779. #ifdef STARPU_USE_SC_HYPERVISOR
  780. sched_ctx->perf_counters = NULL;
  781. #endif
  782. return sched_ctx->id;
  783. }
  784. int fstarpu_sched_ctx_create(int *workerids, int nworkers, const char *sched_ctx_name, void ***_arglist)
  785. {
  786. void **arglist = *_arglist;
  787. int arg_i = 0;
  788. int min_prio_set = 0;
  789. int max_prio_set = 0;
  790. int min_prio = 0;
  791. int max_prio = 0;
  792. int nsms = 0;
  793. int *sub_ctxs = NULL;
  794. int nsub_ctxs = 0;
  795. void *user_data = NULL;
  796. struct starpu_sched_policy *sched_policy = NULL;
  797. unsigned hierarchy_level = 0;
  798. unsigned nesting_sched_ctx = STARPU_NMAX_SCHED_CTXS;
  799. unsigned awake_workers = 0;
  800. void (*init_sched)(unsigned) = NULL;
  801. while (arglist[arg_i] != NULL)
  802. {
  803. const int arg_type = (int)(intptr_t)arglist[arg_i];
  804. if (arg_type == STARPU_SCHED_CTX_POLICY_NAME)
  805. {
  806. arg_i++;
  807. char *policy_name = arglist[arg_i];
  808. struct _starpu_machine_config *config = _starpu_get_machine_config();
  809. sched_policy = _starpu_select_sched_policy(config, policy_name);
  810. }
  811. else if (arg_type == STARPU_SCHED_CTX_POLICY_STRUCT)
  812. {
  813. arg_i++;
  814. sched_policy = arglist[arg_i];
  815. }
  816. else if (arg_type == STARPU_SCHED_CTX_POLICY_MIN_PRIO)
  817. {
  818. arg_i++;
  819. min_prio = *(int *)arglist[arg_i];
  820. min_prio_set = 1;
  821. }
  822. else if (arg_type == STARPU_SCHED_CTX_POLICY_MAX_PRIO)
  823. {
  824. arg_i++;
  825. max_prio = *(int *)arglist[arg_i];
  826. max_prio_set = 1;
  827. }
  828. else if (arg_type == STARPU_SCHED_CTX_HIERARCHY_LEVEL)
  829. {
  830. arg_i++;
  831. int val = *(int *)arglist[arg_i];
  832. STARPU_ASSERT(val >= 0);
  833. hierarchy_level = (unsigned)val;
  834. }
  835. else if (arg_type == STARPU_SCHED_CTX_NESTED)
  836. {
  837. arg_i++;
  838. int val = *(int *)arglist[arg_i];
  839. STARPU_ASSERT(val >= 0);
  840. nesting_sched_ctx = (unsigned)val;
  841. }
  842. else if (arg_type == STARPU_SCHED_CTX_AWAKE_WORKERS)
  843. {
  844. awake_workers = 1;
  845. }
  846. else if (arg_type == STARPU_SCHED_CTX_POLICY_INIT)
  847. {
  848. arg_i++;
  849. init_sched = arglist[arg_i];
  850. }
  851. else if (arg_type == STARPU_SCHED_CTX_USER_DATA)
  852. {
  853. arg_i++;
  854. user_data = arglist[arg_i];
  855. }
  856. else if (arg_type == STARPU_SCHED_CTX_SUB_CTXS)
  857. {
  858. arg_i++;
  859. sub_ctxs = (int*)arglist[arg_i];
  860. arg_i++;
  861. nsub_ctxs = *(int*)arglist[arg_i];
  862. }
  863. else if (arg_type == STARPU_SCHED_CTX_CUDA_NSMS)
  864. {
  865. arg_i++;
  866. nsms = *(int*)arglist[arg_i];
  867. }
  868. else
  869. {
  870. STARPU_ABORT_MSG("Unrecognized argument %d\n", arg_type);
  871. }
  872. arg_i++;
  873. }
  874. if (workerids && nworkers != -1)
  875. {
  876. /* Make sure the user doesn't use invalid worker IDs. */
  877. int num_workers = starpu_worker_get_count();
  878. int i;
  879. for (i = 0; i < nworkers; i++)
  880. {
  881. if (workerids[i] < 0 || workerids[i] >= num_workers)
  882. {
  883. _STARPU_ERROR("Invalid worker ID (%d) specified!\n", workerids[i]);
  884. return STARPU_NMAX_SCHED_CTXS;
  885. }
  886. }
  887. }
  888. struct _starpu_sched_ctx *sched_ctx = NULL;
  889. sched_ctx = _starpu_create_sched_ctx(sched_policy, workerids, nworkers, 0, sched_ctx_name, min_prio_set, min_prio, max_prio_set, max_prio, awake_workers, init_sched, user_data, nsub_ctxs, sub_ctxs, nsms);
  890. sched_ctx->hierarchy_level = hierarchy_level;
  891. sched_ctx->nesting_sched_ctx = nesting_sched_ctx;
  892. _starpu_unlock_mutex_if_prev_locked();
  893. int *added_workerids;
  894. unsigned nw_ctx = starpu_sched_ctx_get_workers_list(sched_ctx->id, &added_workerids);
  895. _starpu_update_workers_with_ctx(added_workerids, nw_ctx, sched_ctx->id);
  896. free(added_workerids);
  897. _starpu_relock_mutex_if_prev_locked();
  898. #ifdef STARPU_USE_SC_HYPERVISOR
  899. sched_ctx->perf_counters = NULL;
  900. #endif
  901. return (int)sched_ctx->id;
  902. }
  903. void starpu_sched_ctx_register_close_callback(unsigned sched_ctx_id, void (*close_callback)(unsigned sched_ctx_id, void* args), void *args)
  904. {
  905. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  906. sched_ctx->close_callback = close_callback;
  907. sched_ctx->close_args = args;
  908. return;
  909. }
  910. #ifdef STARPU_USE_SC_HYPERVISOR
  911. void starpu_sched_ctx_set_perf_counters(unsigned sched_ctx_id, void* perf_counters)
  912. {
  913. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  914. sched_ctx->perf_counters = (struct starpu_sched_ctx_performance_counters *)perf_counters;
  915. return;
  916. }
  917. #endif
  918. /* free all structures for the context */
  919. static void _starpu_delete_sched_ctx(struct _starpu_sched_ctx *sched_ctx)
  920. {
  921. STARPU_ASSERT(sched_ctx->id != STARPU_NMAX_SCHED_CTXS);
  922. struct _starpu_machine_config *config = _starpu_get_machine_config();
  923. int nworkers = config->topology.nworkers;
  924. int w;
  925. for(w = 0; w < nworkers; w++)
  926. {
  927. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx->parallel_sect_mutex[w]);
  928. while (sched_ctx->busy[w]) {
  929. STARPU_PTHREAD_COND_WAIT(&sched_ctx->parallel_sect_cond_busy[w], &sched_ctx->parallel_sect_mutex[w]);
  930. }
  931. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->parallel_sect_mutex[w]);
  932. }
  933. if(sched_ctx->sched_policy)
  934. {
  935. _starpu_deinit_sched_policy(sched_ctx);
  936. free(sched_ctx->sched_policy);
  937. sched_ctx->sched_policy = NULL;
  938. }
  939. else
  940. {
  941. starpu_sched_ctx_delete_worker_collection(sched_ctx->id);
  942. }
  943. if (sched_ctx->perf_arch.devices)
  944. {
  945. free(sched_ctx->perf_arch.devices);
  946. sched_ctx->perf_arch.devices = NULL;
  947. }
  948. STARPU_PTHREAD_MUTEX_DESTROY(&sched_ctx->empty_ctx_mutex);
  949. STARPU_PTHREAD_MUTEX_DESTROY(&sched_ctx->waiting_tasks_mutex);
  950. STARPU_PTHREAD_MUTEX_DESTROY(&sched_ctx->sched_ctx_list_mutex);
  951. sched_ctx->id = STARPU_NMAX_SCHED_CTXS;
  952. #ifdef STARPU_HAVE_HWLOC
  953. hwloc_bitmap_free(sched_ctx->hwloc_workers_set);
  954. #endif //STARPU_HAVE_HWLOC
  955. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx_manag);
  956. config->topology.nsched_ctxs--;
  957. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx_manag);
  958. }
  959. void starpu_sched_ctx_delete(unsigned sched_ctx_id)
  960. {
  961. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  962. STARPU_ASSERT(sched_ctx);
  963. #ifdef STARPU_USE_SC_HYPERVISOR
  964. if (sched_ctx_id != 0 && sched_ctx_id != STARPU_NMAX_SCHED_CTXS && sched_ctx->perf_counters != NULL)
  965. {
  966. _STARPU_TRACE_HYPERVISOR_BEGIN();
  967. sched_ctx->perf_counters->notify_delete_context(sched_ctx_id);
  968. _STARPU_TRACE_HYPERVISOR_END();
  969. }
  970. #endif //STARPU_USE_SC_HYPERVISOR
  971. unsigned inheritor_sched_ctx_id = sched_ctx->inheritor;
  972. struct _starpu_sched_ctx *inheritor_sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx->inheritor);
  973. _starpu_unlock_mutex_if_prev_locked();
  974. STARPU_PTHREAD_RWLOCK_WRLOCK(&changing_ctx_mutex[sched_ctx_id]);
  975. STARPU_ASSERT(sched_ctx->id != STARPU_NMAX_SCHED_CTXS);
  976. int *workerids;
  977. unsigned nworkers_ctx = starpu_sched_ctx_get_workers_list(sched_ctx->id, &workerids);
  978. /*if both of them have all the ressources is pointless*/
  979. /*trying to transfer ressources from one ctx to the other*/
  980. struct _starpu_machine_config *config = _starpu_get_machine_config();
  981. unsigned nworkers = config->topology.nworkers;
  982. if(nworkers_ctx > 0 && inheritor_sched_ctx && inheritor_sched_ctx->id != STARPU_NMAX_SCHED_CTXS &&
  983. !(nworkers_ctx == nworkers && nworkers_ctx == inheritor_sched_ctx->workers->nworkers))
  984. {
  985. starpu_sched_ctx_add_workers(workerids, nworkers_ctx, inheritor_sched_ctx_id);
  986. starpu_sched_ctx_set_priority(workerids, nworkers_ctx, inheritor_sched_ctx_id, 1);
  987. starpu_sched_ctx_set_priority_on_level(workerids, nworkers_ctx, inheritor_sched_ctx_id, 1);
  988. }
  989. if(!_starpu_wait_for_all_tasks_of_sched_ctx(sched_ctx_id))
  990. {
  991. if(!sched_ctx->sched_policy)
  992. _starpu_sched_ctx_wake_up_workers(sched_ctx_id, 0);
  993. /*if btw the mutex release & the mutex lock the context has changed take care to free all
  994. scheduling data before deleting the context */
  995. _starpu_update_workers_without_ctx(workerids, nworkers_ctx, sched_ctx_id, 1);
  996. _starpu_sched_ctx_free_scheduling_data(sched_ctx);
  997. _starpu_delete_sched_ctx(sched_ctx);
  998. }
  999. STARPU_PTHREAD_RWLOCK_UNLOCK(&changing_ctx_mutex[sched_ctx_id]);
  1000. /* workerids is malloc-ed in starpu_sched_ctx_get_workers_list, don't forget to free it when
  1001. you don't use it anymore */
  1002. free(workerids);
  1003. _starpu_relock_mutex_if_prev_locked();
  1004. occupied_sms -= sched_ctx->nsms;
  1005. return;
  1006. }
  1007. /* called after the workers are terminated so we don't have anything else to do but free the memory*/
  1008. void _starpu_delete_all_sched_ctxs()
  1009. {
  1010. unsigned i;
  1011. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  1012. {
  1013. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(i);
  1014. STARPU_PTHREAD_RWLOCK_WRLOCK(&changing_ctx_mutex[i]);
  1015. if(sched_ctx->id != STARPU_NMAX_SCHED_CTXS)
  1016. {
  1017. _starpu_sched_ctx_free_scheduling_data(sched_ctx);
  1018. _starpu_barrier_counter_destroy(&sched_ctx->tasks_barrier);
  1019. _starpu_barrier_counter_destroy(&sched_ctx->ready_tasks_barrier);
  1020. _starpu_delete_sched_ctx(sched_ctx);
  1021. }
  1022. STARPU_PTHREAD_RWLOCK_UNLOCK(&changing_ctx_mutex[i]);
  1023. STARPU_PTHREAD_RWLOCK_DESTROY(&changing_ctx_mutex[i]);
  1024. }
  1025. STARPU_PTHREAD_KEY_DELETE(sched_ctx_key);
  1026. return;
  1027. }
  1028. static void _starpu_check_workers(int *workerids, int nworkers)
  1029. {
  1030. struct _starpu_machine_config *config = _starpu_get_machine_config();
  1031. int nworkers_conf = config->topology.nworkers;
  1032. int i;
  1033. for(i = 0; i < nworkers; i++)
  1034. {
  1035. /* take care the user does not ask for a resource that does not exist */
  1036. STARPU_ASSERT_MSG(workerids[i] >= 0 && workerids[i] <= nworkers_conf, "requested to add workerid = %d, but that is beyond the range 0 to %d", workerids[i], nworkers_conf);
  1037. }
  1038. }
  1039. void _starpu_fetch_tasks_from_empty_ctx_list(struct _starpu_sched_ctx *sched_ctx)
  1040. {
  1041. unsigned unlocked = 0;
  1042. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx->empty_ctx_mutex);
  1043. if(starpu_task_list_empty(&sched_ctx->empty_ctx_tasks))
  1044. {
  1045. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->empty_ctx_mutex);
  1046. return;
  1047. }
  1048. else
  1049. /* you're not suppose to get here if you deleted the context
  1050. so no point in having the mutex locked */
  1051. STARPU_PTHREAD_RWLOCK_UNLOCK(&changing_ctx_mutex[sched_ctx->id]);
  1052. while(!starpu_task_list_empty(&sched_ctx->empty_ctx_tasks))
  1053. {
  1054. if(unlocked)
  1055. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx->empty_ctx_mutex);
  1056. struct starpu_task *old_task = starpu_task_list_pop_back(&sched_ctx->empty_ctx_tasks);
  1057. unlocked = 1;
  1058. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->empty_ctx_mutex);
  1059. if(old_task == &stop_submission_task)
  1060. break;
  1061. int ret = _starpu_push_task_to_workers(old_task);
  1062. /* if we should stop poping from empty ctx tasks */
  1063. if(ret == -EAGAIN) break;
  1064. }
  1065. if(!unlocked)
  1066. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->empty_ctx_mutex);
  1067. /* leave the mutex as it was to avoid pbs in the caller function */
  1068. STARPU_PTHREAD_RWLOCK_RDLOCK(&changing_ctx_mutex[sched_ctx->id]);
  1069. return;
  1070. }
  1071. unsigned _starpu_can_push_task(struct _starpu_sched_ctx *sched_ctx, struct starpu_task *task)
  1072. {
  1073. if(sched_ctx->sched_policy && sched_ctx->sched_policy->simulate_push_task)
  1074. {
  1075. if (window_size == 0.0) return 1;
  1076. STARPU_PTHREAD_RWLOCK_RDLOCK(&changing_ctx_mutex[sched_ctx->id]);
  1077. double expected_end = sched_ctx->sched_policy->simulate_push_task(task);
  1078. STARPU_PTHREAD_RWLOCK_UNLOCK(&changing_ctx_mutex[sched_ctx->id]);
  1079. double expected_len = 0.0;
  1080. if(hyp_actual_start_sample[sched_ctx->id] != 0.0)
  1081. expected_len = expected_end - hyp_actual_start_sample[sched_ctx->id] ;
  1082. else
  1083. {
  1084. printf("%d: sc start is 0.0\n", sched_ctx->id);
  1085. expected_len = expected_end - starpu_timing_now();
  1086. }
  1087. if(expected_len < 0.0)
  1088. printf("exp len negative %lf \n", expected_len);
  1089. expected_len /= 1000000.0;
  1090. // printf("exp_end %lf start %lf expected_len %lf \n", expected_end, hyp_actual_start_sample[sched_ctx->id], expected_len);
  1091. if(expected_len > (window_size + 0.2*window_size))
  1092. return 0;
  1093. }
  1094. return 1;
  1095. }
  1096. void _starpu_fetch_task_from_waiting_list(struct _starpu_sched_ctx *sched_ctx)
  1097. {
  1098. if(starpu_task_list_empty(&sched_ctx->waiting_tasks))
  1099. return;
  1100. struct starpu_task *old_task = starpu_task_list_back(&sched_ctx->waiting_tasks);
  1101. if(_starpu_can_push_task(sched_ctx, old_task))
  1102. {
  1103. old_task = starpu_task_list_pop_back(&sched_ctx->waiting_tasks);
  1104. _starpu_push_task_to_workers(old_task);
  1105. }
  1106. return;
  1107. }
  1108. void _starpu_push_task_to_waiting_list(struct _starpu_sched_ctx *sched_ctx, struct starpu_task *task)
  1109. {
  1110. starpu_task_list_push_front(&sched_ctx->waiting_tasks, task);
  1111. return;
  1112. }
  1113. void starpu_sched_ctx_set_priority_on_level(int* workers_to_add, unsigned nworkers_to_add, unsigned sched_ctx, unsigned priority)
  1114. {
  1115. (void) workers_to_add;
  1116. (void) nworkers_to_add;
  1117. (void) sched_ctx;
  1118. (void) priority;
  1119. /* int w; */
  1120. /* struct _starpu_worker *worker = NULL; */
  1121. /* for(w = 0; w < nworkers_to_add; w++) */
  1122. /* { */
  1123. /* worker = _starpu_get_worker_struct(workers_to_add[w]); */
  1124. /* STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex); */
  1125. /* struct _starpu_sched_ctx_list *l = NULL; */
  1126. /* for (l = worker->sched_ctx_list; l; l = l->next) */
  1127. /* { */
  1128. /* if(l->sched_ctx != STARPU_NMAX_SCHED_CTXS && l->sched_ctx != sched_ctx && */
  1129. /* starpu_sched_ctx_get_hierarchy_level(l->sched_ctx) == starpu_sched_ctx_get_hierarchy_level(sched_ctx)) */
  1130. /* { */
  1131. /* /\* the lock is taken inside the func *\/ */
  1132. /* STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex); */
  1133. /* starpu_sched_ctx_set_priority(&workers_to_add[w], 1, l->sched_ctx, priority); */
  1134. /* STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex); */
  1135. /* } */
  1136. /* } */
  1137. /* STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex); */
  1138. /* } */
  1139. /* return; */
  1140. }
  1141. static void _set_priority_hierarchically(int* workers_to_add, unsigned nworkers_to_add, unsigned sched_ctx, unsigned priority)
  1142. {
  1143. if(starpu_sched_ctx_get_hierarchy_level(sched_ctx) > 0)
  1144. {
  1145. unsigned father = starpu_sched_ctx_get_inheritor(sched_ctx);
  1146. starpu_sched_ctx_set_priority(workers_to_add, nworkers_to_add, father, priority);
  1147. starpu_sched_ctx_set_priority_on_level(workers_to_add, nworkers_to_add, father, priority);
  1148. _set_priority_hierarchically(workers_to_add, nworkers_to_add, father, priority);
  1149. }
  1150. return;
  1151. }
  1152. void starpu_sched_ctx_add_workers(int *workers_to_add, int nworkers_to_add, unsigned sched_ctx_id)
  1153. {
  1154. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1155. _starpu_unlock_mutex_if_prev_locked();
  1156. STARPU_PTHREAD_RWLOCK_WRLOCK(&changing_ctx_mutex[sched_ctx_id]);
  1157. STARPU_ASSERT(workers_to_add != NULL && nworkers_to_add > 0);
  1158. _starpu_check_workers(workers_to_add, nworkers_to_add);
  1159. /* if the context has not already been deleted */
  1160. if(sched_ctx->id != STARPU_NMAX_SCHED_CTXS)
  1161. {
  1162. int added_workers[nworkers_to_add];
  1163. int n_added_workers = 0;
  1164. _starpu_add_workers_to_sched_ctx(sched_ctx, workers_to_add, nworkers_to_add, added_workers, &n_added_workers);
  1165. if(n_added_workers > 0)
  1166. {
  1167. _starpu_update_workers_with_ctx(added_workers, n_added_workers, sched_ctx->id);
  1168. }
  1169. starpu_sched_ctx_set_priority(workers_to_add, nworkers_to_add, sched_ctx_id, 1);
  1170. _set_priority_hierarchically(workers_to_add, nworkers_to_add, sched_ctx_id, 0);
  1171. }
  1172. STARPU_PTHREAD_RWLOCK_UNLOCK(&changing_ctx_mutex[sched_ctx_id]);
  1173. _starpu_relock_mutex_if_prev_locked();
  1174. if(sched_ctx->id != STARPU_NMAX_SCHED_CTXS)
  1175. {
  1176. STARPU_PTHREAD_RWLOCK_RDLOCK(&changing_ctx_mutex[sched_ctx_id]);
  1177. _starpu_fetch_tasks_from_empty_ctx_list(sched_ctx);
  1178. STARPU_PTHREAD_RWLOCK_UNLOCK(&changing_ctx_mutex[sched_ctx_id]);
  1179. }
  1180. return;
  1181. }
  1182. void starpu_sched_ctx_remove_workers(int *workers_to_remove, int nworkers_to_remove, unsigned sched_ctx_id)
  1183. {
  1184. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1185. _starpu_check_workers(workers_to_remove, nworkers_to_remove);
  1186. _starpu_unlock_mutex_if_prev_locked();
  1187. STARPU_PTHREAD_RWLOCK_WRLOCK(&changing_ctx_mutex[sched_ctx_id]);
  1188. /* if the context has not already been deleted */
  1189. if(sched_ctx->id != STARPU_NMAX_SCHED_CTXS)
  1190. {
  1191. int removed_workers[sched_ctx->workers->nworkers];
  1192. int n_removed_workers = 0;
  1193. _starpu_remove_workers_from_sched_ctx(sched_ctx, workers_to_remove, nworkers_to_remove, removed_workers, &n_removed_workers);
  1194. if(n_removed_workers > 0)
  1195. {
  1196. _starpu_update_workers_without_ctx(removed_workers, n_removed_workers, sched_ctx_id, 0);
  1197. starpu_sched_ctx_set_priority(removed_workers, n_removed_workers, sched_ctx_id, 1);
  1198. }
  1199. }
  1200. STARPU_PTHREAD_RWLOCK_UNLOCK(&changing_ctx_mutex[sched_ctx_id]);
  1201. _starpu_relock_mutex_if_prev_locked();
  1202. return;
  1203. }
  1204. int _starpu_nworkers_able_to_execute_task(struct starpu_task *task, struct _starpu_sched_ctx *sched_ctx)
  1205. {
  1206. unsigned nworkers = 0;
  1207. STARPU_PTHREAD_RWLOCK_WRLOCK(&changing_ctx_mutex[sched_ctx->id]);
  1208. struct starpu_worker_collection *workers = sched_ctx->workers;
  1209. struct starpu_sched_ctx_iterator it;
  1210. workers->init_iterator_for_parallel_tasks(workers, &it, task);
  1211. while(workers->has_next(workers, &it))
  1212. {
  1213. unsigned worker = workers->get_next(workers, &it);
  1214. STARPU_ASSERT_MSG(worker < STARPU_NMAXWORKERS, "worker id %d", worker);
  1215. if (starpu_worker_can_execute_task_first_impl(worker, task, NULL))
  1216. nworkers++;
  1217. }
  1218. STARPU_PTHREAD_RWLOCK_UNLOCK(&changing_ctx_mutex[sched_ctx->id]);
  1219. return nworkers;
  1220. }
  1221. /* unused sched_ctx have the id STARPU_NMAX_SCHED_CTXS */
  1222. void _starpu_init_all_sched_ctxs(struct _starpu_machine_config *config)
  1223. {
  1224. STARPU_PTHREAD_KEY_CREATE(&sched_ctx_key, NULL);
  1225. window_size = starpu_get_env_float_default("STARPU_WINDOW_TIME_SIZE", 0.0);
  1226. nobind = starpu_get_env_number("STARPU_WORKERS_NOBIND");
  1227. unsigned i;
  1228. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  1229. config->sched_ctxs[i].id = STARPU_NMAX_SCHED_CTXS;
  1230. return;
  1231. }
  1232. /* sched_ctx aren't necessarly one next to another */
  1233. /* for eg when we remove one its place is free */
  1234. /* when we add new one we reuse its place */
  1235. static unsigned _starpu_get_first_free_sched_ctx(struct _starpu_machine_config *config)
  1236. {
  1237. unsigned i;
  1238. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  1239. if(config->sched_ctxs[i].id == STARPU_NMAX_SCHED_CTXS)
  1240. return i;
  1241. STARPU_ASSERT(0);
  1242. return STARPU_NMAX_SCHED_CTXS;
  1243. }
  1244. int _starpu_wait_for_all_tasks_of_sched_ctx(unsigned sched_ctx_id)
  1245. {
  1246. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1247. STARPU_ASSERT_MSG(_starpu_worker_may_perform_blocking_calls(), "starpu_task_wait_for_all must not be called from a task or callback");
  1248. _starpu_barrier_counter_wait_for_empty_counter(&sched_ctx->tasks_barrier);
  1249. return 0;
  1250. }
  1251. int _starpu_wait_for_n_submitted_tasks_of_sched_ctx(unsigned sched_ctx_id, unsigned n)
  1252. {
  1253. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1254. STARPU_ASSERT_MSG(_starpu_worker_may_perform_blocking_calls(), "starpu_task_wait_for_n_submitted_tasks must not be called from a task or callback");
  1255. return _starpu_barrier_counter_wait_until_counter_reaches_down_to_n(&sched_ctx->tasks_barrier, n);
  1256. }
  1257. void _starpu_decrement_nsubmitted_tasks_of_sched_ctx(unsigned sched_ctx_id)
  1258. {
  1259. struct _starpu_machine_config *config = _starpu_get_machine_config();
  1260. #ifndef STARPU_SANITIZE_THREAD
  1261. if (!config->watchdog_ok)
  1262. config->watchdog_ok = 1;
  1263. #endif
  1264. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1265. int reached = _starpu_barrier_counter_get_reached_start(&sched_ctx->tasks_barrier);
  1266. int finished = reached == 1;
  1267. /* when finished decrementing the tasks if the user signaled he will not submit tasks anymore
  1268. we can move all its workers to the inheritor context */
  1269. if(finished && sched_ctx->inheritor != STARPU_NMAX_SCHED_CTXS)
  1270. {
  1271. STARPU_PTHREAD_MUTEX_LOCK(&finished_submit_mutex);
  1272. if(sched_ctx->finished_submit)
  1273. {
  1274. STARPU_PTHREAD_MUTEX_UNLOCK(&finished_submit_mutex);
  1275. if(sched_ctx->id != STARPU_NMAX_SCHED_CTXS)
  1276. {
  1277. if(sched_ctx->close_callback)
  1278. sched_ctx->close_callback(sched_ctx->id, sched_ctx->close_args);
  1279. int *workerids = NULL;
  1280. unsigned nworkers = starpu_sched_ctx_get_workers_list(sched_ctx->id, &workerids);
  1281. if(nworkers > 0)
  1282. {
  1283. starpu_sched_ctx_add_workers(workerids, nworkers, sched_ctx->inheritor);
  1284. free(workerids);
  1285. }
  1286. }
  1287. _starpu_barrier_counter_decrement_until_empty_counter(&sched_ctx->tasks_barrier, 0.0);
  1288. return;
  1289. }
  1290. STARPU_PTHREAD_MUTEX_UNLOCK(&finished_submit_mutex);
  1291. }
  1292. /* We also need to check for config->submitting = 0 (i.e. the
  1293. * user calle starpu_drivers_request_termination()), in which
  1294. * case we need to set config->running to 0 and wake workers,
  1295. * so they can terminate, just like
  1296. * starpu_drivers_request_termination() does.
  1297. */
  1298. STARPU_PTHREAD_MUTEX_LOCK(&config->submitted_mutex);
  1299. if(config->submitting == 0)
  1300. {
  1301. if(sched_ctx->id != STARPU_NMAX_SCHED_CTXS)
  1302. {
  1303. if(sched_ctx->close_callback)
  1304. sched_ctx->close_callback(sched_ctx->id, sched_ctx->close_args);
  1305. }
  1306. ANNOTATE_HAPPENS_AFTER(&config->running);
  1307. config->running = 0;
  1308. ANNOTATE_HAPPENS_BEFORE(&config->running);
  1309. int s;
  1310. for(s = 0; s < STARPU_NMAX_SCHED_CTXS; s++)
  1311. {
  1312. if(config->sched_ctxs[s].id != STARPU_NMAX_SCHED_CTXS)
  1313. {
  1314. _starpu_check_nsubmitted_tasks_of_sched_ctx(config->sched_ctxs[s].id);
  1315. }
  1316. }
  1317. }
  1318. STARPU_PTHREAD_MUTEX_UNLOCK(&config->submitted_mutex);
  1319. _starpu_barrier_counter_decrement_until_empty_counter(&sched_ctx->tasks_barrier, 0.0);
  1320. return;
  1321. }
  1322. void _starpu_increment_nsubmitted_tasks_of_sched_ctx(unsigned sched_ctx_id)
  1323. {
  1324. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1325. _starpu_barrier_counter_increment(&sched_ctx->tasks_barrier, 0.0);
  1326. }
  1327. int _starpu_get_nsubmitted_tasks_of_sched_ctx(unsigned sched_ctx_id)
  1328. {
  1329. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1330. return _starpu_barrier_counter_get_reached_start(&sched_ctx->tasks_barrier);
  1331. }
  1332. int _starpu_check_nsubmitted_tasks_of_sched_ctx(unsigned sched_ctx_id)
  1333. {
  1334. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1335. return _starpu_barrier_counter_check(&sched_ctx->tasks_barrier);
  1336. }
  1337. unsigned _starpu_increment_nready_tasks_of_sched_ctx(unsigned sched_ctx_id, double ready_flops, struct starpu_task *task)
  1338. {
  1339. unsigned ret = 1;
  1340. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1341. if(!sched_ctx->is_initial_sched)
  1342. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx->waiting_tasks_mutex);
  1343. _starpu_barrier_counter_increment(&sched_ctx->ready_tasks_barrier, ready_flops);
  1344. if(!sched_ctx->is_initial_sched)
  1345. {
  1346. if(!_starpu_can_push_task(sched_ctx, task))
  1347. {
  1348. _starpu_push_task_to_waiting_list(sched_ctx, task);
  1349. ret = 0;
  1350. }
  1351. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->waiting_tasks_mutex);
  1352. }
  1353. return ret;
  1354. }
  1355. void _starpu_decrement_nready_tasks_of_sched_ctx(unsigned sched_ctx_id, double ready_flops)
  1356. {
  1357. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1358. if(!sched_ctx->is_initial_sched)
  1359. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx->waiting_tasks_mutex);
  1360. _starpu_barrier_counter_decrement_until_empty_counter(&sched_ctx->ready_tasks_barrier, ready_flops);
  1361. if(!sched_ctx->is_initial_sched)
  1362. {
  1363. _starpu_fetch_task_from_waiting_list(sched_ctx);
  1364. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->waiting_tasks_mutex);
  1365. }
  1366. }
  1367. int starpu_sched_ctx_get_nready_tasks(unsigned sched_ctx_id)
  1368. {
  1369. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1370. return _starpu_barrier_counter_get_reached_start(&sched_ctx->ready_tasks_barrier);
  1371. }
  1372. double starpu_sched_ctx_get_nready_flops(unsigned sched_ctx_id)
  1373. {
  1374. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1375. return _starpu_barrier_counter_get_reached_flops(&sched_ctx->ready_tasks_barrier);
  1376. }
  1377. int _starpu_wait_for_no_ready_of_sched_ctx(unsigned sched_ctx_id)
  1378. {
  1379. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1380. _starpu_barrier_counter_wait_for_empty_counter(&sched_ctx->ready_tasks_barrier);
  1381. return 0;
  1382. }
  1383. void starpu_sched_ctx_set_context(unsigned *sched_ctx)
  1384. {
  1385. starpu_pthread_setspecific(sched_ctx_key, (void*)sched_ctx);
  1386. }
  1387. unsigned starpu_sched_ctx_get_context()
  1388. {
  1389. unsigned *sched_ctx = (unsigned*)starpu_pthread_getspecific(sched_ctx_key);
  1390. if(sched_ctx == NULL)
  1391. return STARPU_NMAX_SCHED_CTXS;
  1392. STARPU_ASSERT(*sched_ctx < STARPU_NMAX_SCHED_CTXS);
  1393. return *sched_ctx;
  1394. }
  1395. unsigned _starpu_sched_ctx_get_current_context()
  1396. {
  1397. unsigned sched_ctx = starpu_sched_ctx_get_context();
  1398. if (sched_ctx == STARPU_NMAX_SCHED_CTXS)
  1399. return _starpu_get_initial_sched_ctx()->id;
  1400. else
  1401. return sched_ctx;
  1402. }
  1403. void starpu_sched_ctx_notify_hypervisor_exists()
  1404. {
  1405. with_hypervisor = 1;
  1406. int i, j;
  1407. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  1408. {
  1409. hyp_start_sample[i] = starpu_timing_now();
  1410. hyp_start_allow_sample[i] = 0.0;
  1411. for(j = 0; j < STARPU_NMAXWORKERS; j++)
  1412. {
  1413. flops[i][j] = 0.0;
  1414. data_size[i][j] = 0;
  1415. }
  1416. hyp_actual_start_sample[i] = 0.0;
  1417. }
  1418. }
  1419. unsigned starpu_sched_ctx_check_if_hypervisor_exists()
  1420. {
  1421. return with_hypervisor;
  1422. }
  1423. void starpu_sched_ctx_update_start_resizing_sample(unsigned sched_ctx_id, double start_sample)
  1424. {
  1425. hyp_actual_start_sample[sched_ctx_id] = start_sample;
  1426. }
  1427. unsigned _starpu_sched_ctx_allow_hypervisor(unsigned sched_ctx_id)
  1428. {
  1429. (void) sched_ctx_id;
  1430. return 1;
  1431. #if 0
  1432. double now = starpu_timing_now();
  1433. if(hyp_start_allow_sample[sched_ctx_id] > 0.0)
  1434. {
  1435. double allow_sample = (now - hyp_start_allow_sample[sched_ctx_id]) / 1000000.0;
  1436. if(allow_sample < 0.001)
  1437. return 1;
  1438. else
  1439. {
  1440. hyp_start_allow_sample[sched_ctx_id] = 0.0;
  1441. hyp_start_sample[sched_ctx_id] = starpu_timing_now();
  1442. return 0;
  1443. }
  1444. }
  1445. double forbid_sample = (now - hyp_start_sample[sched_ctx_id]) / 1000000.0;
  1446. if(forbid_sample > 0.01)
  1447. {
  1448. // hyp_start_sample[sched_ctx_id] = starpu_timing_now();
  1449. hyp_start_allow_sample[sched_ctx_id] = starpu_timing_now();
  1450. return 1;
  1451. }
  1452. return 0;
  1453. #endif
  1454. }
  1455. void starpu_sched_ctx_set_policy_data(unsigned sched_ctx_id, void* policy_data)
  1456. {
  1457. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1458. sched_ctx->policy_data = policy_data;
  1459. }
  1460. void* starpu_sched_ctx_get_policy_data(unsigned sched_ctx_id)
  1461. {
  1462. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1463. return sched_ctx->policy_data;
  1464. }
  1465. struct starpu_sched_policy *starpu_sched_ctx_get_sched_policy(unsigned sched_ctx_id)
  1466. {
  1467. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1468. return sched_ctx->sched_policy;
  1469. }
  1470. struct starpu_worker_collection* starpu_sched_ctx_create_worker_collection(unsigned sched_ctx_id, enum starpu_worker_collection_type worker_collection_type)
  1471. {
  1472. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1473. _STARPU_MALLOC(sched_ctx->workers, sizeof(struct starpu_worker_collection));
  1474. switch(worker_collection_type)
  1475. {
  1476. #ifdef STARPU_HAVE_HWLOC
  1477. case STARPU_WORKER_TREE:
  1478. sched_ctx->workers->has_next = worker_tree.has_next;
  1479. sched_ctx->workers->get_next = worker_tree.get_next;
  1480. sched_ctx->workers->add = worker_tree.add;
  1481. sched_ctx->workers->remove = worker_tree.remove;
  1482. sched_ctx->workers->init = worker_tree.init;
  1483. sched_ctx->workers->deinit = worker_tree.deinit;
  1484. sched_ctx->workers->init_iterator = worker_tree.init_iterator;
  1485. sched_ctx->workers->init_iterator_for_parallel_tasks = worker_tree.init_iterator_for_parallel_tasks;
  1486. sched_ctx->workers->type = STARPU_WORKER_TREE;
  1487. break;
  1488. #endif
  1489. // case STARPU_WORKER_LIST:
  1490. default:
  1491. sched_ctx->workers->has_next = worker_list.has_next;
  1492. sched_ctx->workers->get_next = worker_list.get_next;
  1493. sched_ctx->workers->add = worker_list.add;
  1494. sched_ctx->workers->remove = worker_list.remove;
  1495. sched_ctx->workers->init = worker_list.init;
  1496. sched_ctx->workers->deinit = worker_list.deinit;
  1497. sched_ctx->workers->init_iterator = worker_list.init_iterator;
  1498. sched_ctx->workers->init_iterator_for_parallel_tasks = worker_list.init_iterator_for_parallel_tasks;
  1499. sched_ctx->workers->type = STARPU_WORKER_LIST;
  1500. break;
  1501. }
  1502. /* construct the collection of workers(list/tree/etc.) */
  1503. sched_ctx->workers->init(sched_ctx->workers);
  1504. return sched_ctx->workers;
  1505. }
  1506. void starpu_sched_ctx_display_workers(unsigned sched_ctx_id, FILE *f)
  1507. {
  1508. int *workerids = NULL;
  1509. unsigned nworkers;
  1510. unsigned i;
  1511. nworkers = starpu_sched_ctx_get_workers_list(sched_ctx_id, &workerids);
  1512. fprintf(f, "[sched_ctx %u]: %u worker%s\n", sched_ctx_id, nworkers, nworkers>1?"s":"");
  1513. for (i = 0; i < nworkers; i++)
  1514. {
  1515. char name[256];
  1516. starpu_worker_get_name(workerids[i], name, 256);
  1517. fprintf(f, "\t\t%s\n", name);
  1518. }
  1519. free(workerids);
  1520. }
  1521. unsigned starpu_sched_ctx_get_workers_list_raw(unsigned sched_ctx_id, int **workerids)
  1522. {
  1523. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1524. *workerids = sched_ctx->workers->workerids;
  1525. return sched_ctx->workers->nworkers;
  1526. }
  1527. unsigned starpu_sched_ctx_get_workers_list(unsigned sched_ctx_id, int **workerids)
  1528. {
  1529. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1530. struct starpu_worker_collection *workers = sched_ctx->workers;
  1531. unsigned nworkers = 0;
  1532. struct starpu_sched_ctx_iterator it;
  1533. if(!workers) return 0;
  1534. _STARPU_MALLOC(*workerids, workers->nworkers*sizeof(int));
  1535. workers->init_iterator(workers, &it);
  1536. while(workers->has_next(workers, &it))
  1537. {
  1538. int worker = workers->get_next(workers, &it);
  1539. (*workerids)[nworkers++] = worker;
  1540. }
  1541. return nworkers;
  1542. }
  1543. void starpu_sched_ctx_delete_worker_collection(unsigned sched_ctx_id)
  1544. {
  1545. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1546. sched_ctx->workers->deinit(sched_ctx->workers);
  1547. free(sched_ctx->workers);
  1548. sched_ctx->workers = NULL;
  1549. }
  1550. struct starpu_worker_collection* starpu_sched_ctx_get_worker_collection(unsigned sched_ctx_id)
  1551. {
  1552. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1553. return sched_ctx->workers;
  1554. }
  1555. int _starpu_get_workers_of_sched_ctx(unsigned sched_ctx_id, int *pus, enum starpu_worker_archtype arch)
  1556. {
  1557. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1558. struct starpu_worker_collection *workers = sched_ctx->workers;
  1559. int npus = 0;
  1560. struct starpu_sched_ctx_iterator it;
  1561. workers->init_iterator(workers, &it);
  1562. while(workers->has_next(workers, &it))
  1563. {
  1564. int worker = workers->get_next(workers, &it);
  1565. enum starpu_worker_archtype curr_arch = starpu_worker_get_type(worker);
  1566. if(curr_arch == arch || arch == STARPU_ANY_WORKER)
  1567. pus[npus++] = worker;
  1568. }
  1569. return npus;
  1570. }
  1571. starpu_pthread_rwlock_t* _starpu_sched_ctx_get_changing_ctx_mutex(unsigned sched_ctx_id)
  1572. {
  1573. return &changing_ctx_mutex[sched_ctx_id];
  1574. }
  1575. unsigned starpu_sched_ctx_get_nworkers(unsigned sched_ctx_id)
  1576. {
  1577. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1578. if(sched_ctx != NULL)
  1579. return sched_ctx->workers->nworkers;
  1580. else
  1581. return 0;
  1582. }
  1583. unsigned starpu_sched_ctx_get_nshared_workers(unsigned sched_ctx_id, unsigned sched_ctx_id2)
  1584. {
  1585. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1586. struct _starpu_sched_ctx *sched_ctx2 = _starpu_get_sched_ctx_struct(sched_ctx_id2);
  1587. struct starpu_worker_collection *workers = sched_ctx->workers;
  1588. struct starpu_worker_collection *workers2 = sched_ctx2->workers;
  1589. int shared_workers = 0;
  1590. struct starpu_sched_ctx_iterator it1, it2;
  1591. workers->init_iterator(workers, &it1);
  1592. workers2->init_iterator(workers2, &it2);
  1593. while(workers->has_next(workers, &it1))
  1594. {
  1595. int worker = workers->get_next(workers, &it1);
  1596. while(workers2->has_next(workers2, &it2))
  1597. {
  1598. int worker2 = workers2->get_next(workers2, &it2);
  1599. if(worker == worker2)
  1600. shared_workers++;
  1601. }
  1602. }
  1603. return shared_workers;
  1604. }
  1605. unsigned starpu_sched_ctx_contains_worker(int workerid, unsigned sched_ctx_id)
  1606. {
  1607. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1608. struct starpu_worker_collection *workers = sched_ctx->workers;
  1609. if(workers)
  1610. {
  1611. unsigned i;
  1612. for (i = 0; i < workers->nworkers; i++)
  1613. if (workerid == workers->workerids[i])
  1614. return 1;
  1615. }
  1616. return 0;
  1617. }
  1618. unsigned starpu_sched_ctx_contains_type_of_worker(enum starpu_worker_archtype arch, unsigned sched_ctx_id)
  1619. {
  1620. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sched_ctx_id);
  1621. unsigned i;
  1622. for (i = 0; i < workers->nworkers; i++)
  1623. {
  1624. int worker = workers->workerids[i];
  1625. enum starpu_worker_archtype curr_arch = starpu_worker_get_type(worker);
  1626. if(curr_arch == arch)
  1627. return 1;
  1628. }
  1629. return 0;
  1630. }
  1631. unsigned _starpu_worker_belongs_to_a_sched_ctx(int workerid, unsigned sched_ctx_id)
  1632. {
  1633. struct _starpu_machine_config *config = _starpu_get_machine_config();
  1634. int i;
  1635. for(i = 0; i < STARPU_NMAX_SCHED_CTXS; i++)
  1636. {
  1637. struct _starpu_sched_ctx *sched_ctx = &config->sched_ctxs[i];
  1638. if(sched_ctx && sched_ctx->id != STARPU_NMAX_SCHED_CTXS && sched_ctx->id != sched_ctx_id)
  1639. if(starpu_sched_ctx_contains_worker(workerid, sched_ctx->id))
  1640. return 1;
  1641. }
  1642. return 0;
  1643. }
  1644. unsigned starpu_sched_ctx_worker_get_id(unsigned sched_ctx_id)
  1645. {
  1646. int workerid = starpu_worker_get_id();
  1647. if(workerid != -1)
  1648. if(starpu_sched_ctx_contains_worker(workerid, sched_ctx_id))
  1649. return workerid;
  1650. return -1;
  1651. }
  1652. unsigned starpu_sched_ctx_get_ctx_for_task(struct starpu_task *task)
  1653. {
  1654. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(task->sched_ctx);
  1655. unsigned ret_sched_ctx = task->sched_ctx;
  1656. if (task->possibly_parallel && !sched_ctx->sched_policy
  1657. && sched_ctx->nesting_sched_ctx != STARPU_NMAX_SCHED_CTXS)
  1658. ret_sched_ctx = sched_ctx->nesting_sched_ctx;
  1659. return ret_sched_ctx;
  1660. }
  1661. unsigned starpu_sched_ctx_overlapping_ctxs_on_worker(int workerid)
  1662. {
  1663. struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
  1664. return worker->nsched_ctxs > 1;
  1665. }
  1666. void starpu_sched_ctx_set_inheritor(unsigned sched_ctx_id, unsigned inheritor)
  1667. {
  1668. STARPU_ASSERT(sched_ctx_id < STARPU_NMAX_SCHED_CTXS);
  1669. STARPU_ASSERT(inheritor < STARPU_NMAX_SCHED_CTXS);
  1670. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1671. sched_ctx->inheritor = inheritor;
  1672. return;
  1673. }
  1674. unsigned starpu_sched_ctx_get_inheritor(unsigned sched_ctx_id)
  1675. {
  1676. STARPU_ASSERT(sched_ctx_id < STARPU_NMAX_SCHED_CTXS);
  1677. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1678. return sched_ctx->inheritor;
  1679. }
  1680. unsigned starpu_sched_ctx_get_hierarchy_level(unsigned sched_ctx_id)
  1681. {
  1682. STARPU_ASSERT(sched_ctx_id < STARPU_NMAX_SCHED_CTXS);
  1683. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1684. return sched_ctx->hierarchy_level;
  1685. }
  1686. void starpu_sched_ctx_finished_submit(unsigned sched_ctx_id)
  1687. {
  1688. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1689. STARPU_PTHREAD_MUTEX_LOCK(&finished_submit_mutex);
  1690. sched_ctx->finished_submit = 1;
  1691. STARPU_PTHREAD_MUTEX_UNLOCK(&finished_submit_mutex);
  1692. return;
  1693. }
  1694. #ifdef STARPU_USE_SC_HYPERVISOR
  1695. void _starpu_sched_ctx_post_exec_task_cb(int workerid, struct starpu_task *task, size_t data_size2, uint32_t footprint)
  1696. {
  1697. if (workerid < 0)
  1698. return;
  1699. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(task->sched_ctx);
  1700. if(sched_ctx != NULL && task->sched_ctx != _starpu_get_initial_sched_ctx()->id &&
  1701. task->sched_ctx != STARPU_NMAX_SCHED_CTXS && sched_ctx->perf_counters != NULL)
  1702. {
  1703. flops[task->sched_ctx][workerid] += task->flops;
  1704. data_size[task->sched_ctx][workerid] += data_size2;
  1705. if(_starpu_sched_ctx_allow_hypervisor(sched_ctx->id) || task->hypervisor_tag > 0)
  1706. {
  1707. _STARPU_TRACE_HYPERVISOR_BEGIN();
  1708. sched_ctx->perf_counters->notify_post_exec_task(task, data_size[task->sched_ctx][workerid], footprint,
  1709. task->hypervisor_tag, flops[task->sched_ctx][workerid]);
  1710. _STARPU_TRACE_HYPERVISOR_END();
  1711. flops[task->sched_ctx][workerid] = 0.0;
  1712. data_size[task->sched_ctx][workerid] = 0;
  1713. }
  1714. }
  1715. }
  1716. void starpu_sched_ctx_call_pushed_task_cb(int workerid, unsigned sched_ctx_id)
  1717. {
  1718. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1719. if(sched_ctx != NULL && sched_ctx_id != _starpu_get_initial_sched_ctx()->id && sched_ctx_id != STARPU_NMAX_SCHED_CTXS
  1720. && sched_ctx->perf_counters != NULL && _starpu_sched_ctx_allow_hypervisor(sched_ctx_id))
  1721. {
  1722. _STARPU_TRACE_HYPERVISOR_BEGIN();
  1723. sched_ctx->perf_counters->notify_pushed_task(sched_ctx_id, workerid);
  1724. _STARPU_TRACE_HYPERVISOR_END();
  1725. }
  1726. }
  1727. #endif //STARPU_USE_SC_HYPERVISOR
  1728. int starpu_sched_get_min_priority(void)
  1729. {
  1730. return starpu_sched_ctx_get_min_priority(_starpu_sched_ctx_get_current_context());
  1731. }
  1732. int starpu_sched_get_max_priority(void)
  1733. {
  1734. return starpu_sched_ctx_get_max_priority(_starpu_sched_ctx_get_current_context());
  1735. }
  1736. int starpu_sched_set_min_priority(int min_prio)
  1737. {
  1738. return starpu_sched_ctx_set_min_priority(_starpu_sched_ctx_get_current_context(), min_prio);
  1739. }
  1740. int starpu_sched_set_max_priority(int max_prio)
  1741. {
  1742. return starpu_sched_ctx_set_max_priority(_starpu_sched_ctx_get_current_context(), max_prio);
  1743. }
  1744. int starpu_sched_ctx_get_min_priority(unsigned sched_ctx_id)
  1745. {
  1746. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1747. return sched_ctx->min_priority;
  1748. }
  1749. int starpu_sched_ctx_get_max_priority(unsigned sched_ctx_id)
  1750. {
  1751. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1752. return sched_ctx->max_priority;
  1753. }
  1754. int starpu_sched_ctx_set_min_priority(unsigned sched_ctx_id, int min_prio)
  1755. {
  1756. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1757. sched_ctx->min_priority = min_prio;
  1758. return 0;
  1759. }
  1760. int starpu_sched_ctx_set_max_priority(unsigned sched_ctx_id, int max_prio)
  1761. {
  1762. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1763. sched_ctx->max_priority = max_prio;
  1764. return 0;
  1765. }
  1766. int starpu_sched_ctx_min_priority_is_set(unsigned sched_ctx_id)
  1767. {
  1768. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1769. return sched_ctx->min_priority_is_set;
  1770. }
  1771. int starpu_sched_ctx_max_priority_is_set(unsigned sched_ctx_id)
  1772. {
  1773. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1774. return sched_ctx->max_priority_is_set;
  1775. }
  1776. void starpu_sched_ctx_set_priority(int *workers, int nworkers, unsigned sched_ctx_id, unsigned priority)
  1777. {
  1778. if(nworkers != -1)
  1779. {
  1780. int w;
  1781. struct _starpu_worker *worker = NULL;
  1782. for(w = 0; w < nworkers; w++)
  1783. {
  1784. worker = _starpu_get_worker_struct(workers[w]);
  1785. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
  1786. _starpu_sched_ctx_list_move(&worker->sched_ctx_list, sched_ctx_id, priority);
  1787. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
  1788. }
  1789. }
  1790. return;
  1791. }
  1792. unsigned starpu_sched_ctx_get_priority(int workerid, unsigned sched_ctx_id)
  1793. {
  1794. struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
  1795. return _starpu_sched_ctx_elt_get_priority(worker->sched_ctx_list, sched_ctx_id);
  1796. }
  1797. unsigned _starpu_sched_ctx_last_worker_awake(struct _starpu_worker *worker)
  1798. {
  1799. struct _starpu_sched_ctx_list_iterator list_it;
  1800. _starpu_sched_ctx_list_iterator_init(worker->sched_ctx_list, &list_it);
  1801. while (_starpu_sched_ctx_list_iterator_has_next(&list_it))
  1802. {
  1803. struct _starpu_sched_ctx_elt *e = _starpu_sched_ctx_list_iterator_get_next(&list_it);
  1804. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(e->sched_ctx);
  1805. unsigned last_worker_awake = 1;
  1806. struct starpu_worker_collection *workers = sched_ctx->workers;
  1807. struct starpu_sched_ctx_iterator it;
  1808. workers->init_iterator(workers, &it);
  1809. while(workers->has_next(workers, &it))
  1810. {
  1811. int workerid = workers->get_next(workers, &it);
  1812. if(workerid != worker->workerid && _starpu_worker_get_status(workerid) != STATUS_SLEEPING)
  1813. {
  1814. last_worker_awake = 0;
  1815. break;
  1816. }
  1817. }
  1818. if(last_worker_awake)
  1819. return 1;
  1820. }
  1821. return 0;
  1822. }
  1823. void starpu_sched_ctx_bind_current_thread_to_cpuid(unsigned cpuid)
  1824. {
  1825. _starpu_bind_thread_on_cpu(_starpu_get_machine_config(), cpuid, STARPU_NOWORKERID);
  1826. }
  1827. unsigned starpu_sched_ctx_worker_is_master_for_child_ctx(int workerid, unsigned sched_ctx_id)
  1828. {
  1829. if (_starpu_get_nsched_ctxs() <= 1)
  1830. return STARPU_NMAX_SCHED_CTXS;
  1831. struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
  1832. struct _starpu_sched_ctx_list_iterator list_it;
  1833. _starpu_sched_ctx_list_iterator_init(worker->sched_ctx_list, &list_it);
  1834. while (_starpu_sched_ctx_list_iterator_has_next(&list_it))
  1835. {
  1836. struct _starpu_sched_ctx_elt *e = _starpu_sched_ctx_list_iterator_get_next(&list_it);
  1837. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(e->sched_ctx);
  1838. if(sched_ctx-> main_master == workerid && sched_ctx->nesting_sched_ctx == sched_ctx_id)
  1839. return sched_ctx->id;
  1840. }
  1841. return STARPU_NMAX_SCHED_CTXS;
  1842. }
  1843. unsigned starpu_sched_ctx_master_get_context(int masterid)
  1844. {
  1845. struct _starpu_worker *worker = _starpu_get_worker_struct(masterid);
  1846. struct _starpu_sched_ctx_list_iterator list_it;
  1847. _starpu_sched_ctx_list_iterator_init(worker->sched_ctx_list, &list_it);
  1848. while (_starpu_sched_ctx_list_iterator_has_next(&list_it))
  1849. {
  1850. struct _starpu_sched_ctx_elt *e = _starpu_sched_ctx_list_iterator_get_next(&list_it);
  1851. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(e->sched_ctx);
  1852. if(sched_ctx->main_master == masterid)
  1853. return sched_ctx->id;
  1854. }
  1855. return STARPU_NMAX_SCHED_CTXS;
  1856. }
  1857. struct _starpu_sched_ctx *__starpu_sched_ctx_get_sched_ctx_for_worker_and_job(struct _starpu_worker *worker, struct _starpu_job *j)
  1858. {
  1859. struct _starpu_sched_ctx_list_iterator list_it;
  1860. _starpu_sched_ctx_list_iterator_init(worker->sched_ctx_list, &list_it);
  1861. while (_starpu_sched_ctx_list_iterator_has_next(&list_it))
  1862. {
  1863. struct _starpu_sched_ctx_elt *e = _starpu_sched_ctx_list_iterator_get_next(&list_it);
  1864. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(e->sched_ctx);
  1865. if (j->task->sched_ctx == sched_ctx->id)
  1866. return sched_ctx;
  1867. }
  1868. return NULL;
  1869. }
  1870. void starpu_sched_ctx_revert_task_counters(unsigned sched_ctx_id, double ready_flops)
  1871. {
  1872. _starpu_decrement_nsubmitted_tasks_of_sched_ctx(sched_ctx_id);
  1873. _starpu_decrement_nready_tasks_of_sched_ctx(sched_ctx_id, ready_flops);
  1874. }
  1875. void starpu_sched_ctx_move_task_to_ctx(struct starpu_task *task, unsigned sched_ctx, unsigned manage_mutex,
  1876. unsigned with_repush)
  1877. {
  1878. /* TODO: make something cleaner which differentiates between calls
  1879. from push or pop (have mutex or not) and from another worker or not */
  1880. int workerid = starpu_worker_get_id();
  1881. struct _starpu_worker *worker = NULL;
  1882. if(workerid != -1 && manage_mutex)
  1883. {
  1884. worker = _starpu_get_worker_struct(workerid);
  1885. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
  1886. }
  1887. task->sched_ctx = sched_ctx;
  1888. struct _starpu_job *j = _starpu_get_job_associated_to_task(task);
  1889. _starpu_increment_nsubmitted_tasks_of_sched_ctx(j->task->sched_ctx);
  1890. if(with_repush)
  1891. _starpu_repush_task(j);
  1892. else
  1893. _starpu_increment_nready_tasks_of_sched_ctx(j->task->sched_ctx, j->task->flops, j->task);
  1894. if(workerid != -1 && manage_mutex)
  1895. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
  1896. }
  1897. void starpu_sched_ctx_list_task_counters_increment(unsigned sched_ctx_id, int workerid)
  1898. {
  1899. /* Note : often we don't have any sched_mutex taken here but we
  1900. should, so take it */
  1901. struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
  1902. if (worker->nsched_ctxs > 1)
  1903. {
  1904. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
  1905. _starpu_sched_ctx_list_push_event(worker->sched_ctx_list, sched_ctx_id);
  1906. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
  1907. }
  1908. }
  1909. void starpu_sched_ctx_list_task_counters_decrement(unsigned sched_ctx_id, int workerid)
  1910. {
  1911. struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
  1912. if (worker->nsched_ctxs > 1)
  1913. _starpu_sched_ctx_list_pop_event(worker->sched_ctx_list, sched_ctx_id);
  1914. }
  1915. void starpu_sched_ctx_list_task_counters_reset(unsigned sched_ctx_id, int workerid)
  1916. {
  1917. struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
  1918. if (worker->nsched_ctxs > 1)
  1919. _starpu_sched_ctx_list_pop_all_event(worker->sched_ctx_list, sched_ctx_id);
  1920. }
  1921. void starpu_sched_ctx_list_task_counters_increment_all(struct starpu_task *task, unsigned sched_ctx_id)
  1922. {
  1923. /* Note that with 1 ctx we will default to the global context,
  1924. hence our counters are useless */
  1925. if (_starpu_get_nsched_ctxs() > 1)
  1926. {
  1927. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1928. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sched_ctx_id);
  1929. struct starpu_sched_ctx_iterator it;
  1930. workers->init_iterator_for_parallel_tasks(workers, &it, task);
  1931. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx->sched_ctx_list_mutex);
  1932. while(workers->has_next(workers, &it))
  1933. {
  1934. int worker = workers->get_next(workers, &it);
  1935. starpu_sched_ctx_list_task_counters_increment(sched_ctx_id, worker);
  1936. }
  1937. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->sched_ctx_list_mutex);
  1938. }
  1939. }
  1940. void starpu_sched_ctx_list_task_counters_decrement_all(struct starpu_task *task, unsigned sched_ctx_id)
  1941. {
  1942. if (_starpu_get_nsched_ctxs() > 1)
  1943. {
  1944. int curr_workerid = starpu_worker_get_id();
  1945. struct _starpu_worker *curr_worker_str = NULL, *worker_str;
  1946. if(curr_workerid != -1)
  1947. {
  1948. curr_worker_str = _starpu_get_worker_struct(curr_workerid);
  1949. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&curr_worker_str->sched_mutex);
  1950. }
  1951. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1952. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sched_ctx_id);
  1953. struct starpu_sched_ctx_iterator it;
  1954. workers->init_iterator_for_parallel_tasks(workers, &it, task);
  1955. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx->sched_ctx_list_mutex);
  1956. while(workers->has_next(workers, &it))
  1957. {
  1958. int worker = workers->get_next(workers, &it);
  1959. worker_str = _starpu_get_worker_struct(worker);
  1960. if (worker_str->nsched_ctxs > 1)
  1961. {
  1962. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker_str->sched_mutex);
  1963. starpu_sched_ctx_list_task_counters_decrement(sched_ctx_id, worker);
  1964. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker_str->sched_mutex);
  1965. }
  1966. }
  1967. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->sched_ctx_list_mutex);
  1968. if(curr_workerid != -1)
  1969. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&curr_worker_str->sched_mutex);
  1970. }
  1971. }
  1972. void starpu_sched_ctx_list_task_counters_reset_all(struct starpu_task *task, unsigned sched_ctx_id)
  1973. {
  1974. if (_starpu_get_nsched_ctxs() > 1)
  1975. {
  1976. int curr_workerid = starpu_worker_get_id();
  1977. struct _starpu_worker *curr_worker_str = NULL, *worker_str;
  1978. if(curr_workerid != -1)
  1979. {
  1980. curr_worker_str = _starpu_get_worker_struct(curr_workerid);
  1981. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&curr_worker_str->sched_mutex);
  1982. }
  1983. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  1984. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sched_ctx_id);
  1985. struct starpu_sched_ctx_iterator it;
  1986. workers->init_iterator_for_parallel_tasks(workers, &it, task);
  1987. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx->sched_ctx_list_mutex);
  1988. while(workers->has_next(workers, &it))
  1989. {
  1990. int worker = workers->get_next(workers, &it);
  1991. worker_str = _starpu_get_worker_struct(worker);
  1992. if (worker_str->nsched_ctxs > 1)
  1993. {
  1994. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker_str->sched_mutex);
  1995. starpu_sched_ctx_list_task_counters_reset(sched_ctx_id, worker);
  1996. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker_str->sched_mutex);
  1997. }
  1998. }
  1999. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->sched_ctx_list_mutex);
  2000. if(curr_workerid != -1)
  2001. STARPU_PTHREAD_MUTEX_LOCK_SCHED(&curr_worker_str->sched_mutex);
  2002. }
  2003. }
  2004. static unsigned _worker_sleeping_in_other_ctx(unsigned sched_ctx_id, int workerid)
  2005. {
  2006. int s;
  2007. for(s = 0; s < STARPU_NMAX_SCHED_CTXS; s++)
  2008. {
  2009. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(s);
  2010. if(sched_ctx && sched_ctx->id > 0 && sched_ctx->id < STARPU_NMAX_SCHED_CTXS && sched_ctx->id != sched_ctx_id)
  2011. {
  2012. if(sched_ctx->parallel_sect[workerid])
  2013. return 1;
  2014. }
  2015. }
  2016. return 0;
  2017. }
  2018. void _starpu_sched_ctx_signal_worker_blocked(unsigned sched_ctx_id, int workerid)
  2019. {
  2020. struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
  2021. worker->blocked = 1;
  2022. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2023. sched_ctx->sleeping[workerid] = 1;
  2024. sem_post(&sched_ctx->fall_asleep_sem[sched_ctx->main_master]);
  2025. return;
  2026. }
  2027. void _starpu_sched_ctx_signal_worker_woke_up(unsigned sched_ctx_id, int workerid)
  2028. {
  2029. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2030. sem_post(&sched_ctx->wake_up_sem[sched_ctx->main_master]);
  2031. sched_ctx->sleeping[workerid] = 0;
  2032. struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
  2033. worker->blocked = 0;
  2034. return;
  2035. }
  2036. static void _starpu_sched_ctx_put_workers_to_sleep(unsigned sched_ctx_id, unsigned all)
  2037. {
  2038. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2039. int current_worker_id = starpu_worker_get_id();
  2040. int master, temp_master = 0;
  2041. struct starpu_worker_collection *workers = sched_ctx->workers;
  2042. struct starpu_sched_ctx_iterator it;
  2043. unsigned sleeping[workers->nworkers];
  2044. int workers_count = 0;
  2045. /* temporarily put a master if needed */
  2046. if (sched_ctx->main_master == -1)
  2047. {
  2048. _starpu_sched_ctx_put_new_master(sched_ctx_id);
  2049. temp_master = 1;
  2050. }
  2051. master = sched_ctx->main_master;
  2052. workers->init_iterator(workers, &it);
  2053. while(workers->has_next(workers, &it))
  2054. {
  2055. int workerid = workers->get_next(workers, &it);
  2056. sleeping[workers_count] = _worker_sleeping_in_other_ctx(sched_ctx_id, workerid);
  2057. if(starpu_worker_get_type(workerid) == STARPU_CPU_WORKER
  2058. && !sched_ctx->parallel_sect[workerid] && (workerid != master || all))
  2059. {
  2060. if (current_worker_id == -1 || workerid != current_worker_id)
  2061. {
  2062. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx->parallel_sect_mutex[workerid]);
  2063. sched_ctx->parallel_sect[workerid] = 1;
  2064. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->parallel_sect_mutex[workerid]);
  2065. }
  2066. }
  2067. workers_count++;
  2068. }
  2069. workers_count = 0;
  2070. workers->init_iterator(workers, &it);
  2071. while(workers->has_next(workers, &it))
  2072. {
  2073. int workerid = workers->get_next(workers, &it);
  2074. if(starpu_worker_get_type(workerid) == STARPU_CPU_WORKER
  2075. && (workerid != master || all)
  2076. && (current_worker_id == -1 || workerid != current_worker_id)
  2077. && !sleeping[workers_count])
  2078. {
  2079. sem_wait(&sched_ctx->fall_asleep_sem[master]);
  2080. }
  2081. workers_count++;
  2082. }
  2083. if (temp_master)
  2084. sched_ctx->main_master = -1;
  2085. return;
  2086. }
  2087. static void _starpu_sched_ctx_wake_up_workers(unsigned sched_ctx_id, unsigned all)
  2088. {
  2089. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2090. int current_worker_id = starpu_worker_get_id();
  2091. int master, temp_master = 0;
  2092. struct starpu_worker_collection *workers = sched_ctx->workers;
  2093. struct starpu_sched_ctx_iterator it;
  2094. /* temporarily put a master if needed */
  2095. if (sched_ctx->main_master == -1)
  2096. {
  2097. _starpu_sched_ctx_put_new_master(sched_ctx_id);
  2098. temp_master = 1;
  2099. }
  2100. master = sched_ctx->main_master;
  2101. workers->init_iterator(workers, &it);
  2102. while(workers->has_next(workers, &it))
  2103. {
  2104. int workerid = workers->get_next(workers, &it);
  2105. if(starpu_worker_get_type(workerid) == STARPU_CPU_WORKER
  2106. && sched_ctx->parallel_sect[workerid] && (workerid != master || all))
  2107. {
  2108. if((current_worker_id == -1 || workerid != current_worker_id) && sched_ctx->sleeping[workerid])
  2109. {
  2110. STARPU_PTHREAD_MUTEX_LOCK(&sched_ctx->parallel_sect_mutex[workerid]);
  2111. STARPU_PTHREAD_COND_SIGNAL(&sched_ctx->parallel_sect_cond[workerid]);
  2112. STARPU_PTHREAD_MUTEX_UNLOCK(&sched_ctx->parallel_sect_mutex[workerid]);
  2113. sem_wait(&sched_ctx->wake_up_sem[master]);
  2114. }
  2115. else
  2116. sched_ctx->parallel_sect[workerid] = 0;
  2117. }
  2118. }
  2119. if (temp_master)
  2120. sched_ctx->main_master = -1;
  2121. return;
  2122. }
  2123. void* starpu_sched_ctx_exec_parallel_code(void* (*func)(void*), void* param, unsigned sched_ctx_id)
  2124. {
  2125. _starpu_sched_ctx_put_workers_to_sleep(sched_ctx_id, 1);
  2126. /* execute parallel code */
  2127. void* ret = func(param);
  2128. /* wake up starpu workers */
  2129. _starpu_sched_ctx_wake_up_workers(sched_ctx_id, 1);
  2130. return ret;
  2131. }
  2132. static void _starpu_sched_ctx_update_parallel_workers_with(unsigned sched_ctx_id)
  2133. {
  2134. struct _starpu_sched_ctx * sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2135. if(sched_ctx->sched_policy)
  2136. return;
  2137. _starpu_sched_ctx_put_new_master(sched_ctx_id);
  2138. if(!sched_ctx->awake_workers)
  2139. {
  2140. _starpu_sched_ctx_put_workers_to_sleep(sched_ctx_id, 0);
  2141. }
  2142. }
  2143. static void _starpu_sched_ctx_update_parallel_workers_without(unsigned sched_ctx_id)
  2144. {
  2145. struct _starpu_sched_ctx * sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2146. if(sched_ctx->sched_policy)
  2147. return;
  2148. _starpu_sched_ctx_put_new_master(sched_ctx_id);
  2149. if(!sched_ctx->awake_workers)
  2150. {
  2151. _starpu_sched_ctx_wake_up_workers(sched_ctx_id, 0);
  2152. }
  2153. }
  2154. void starpu_sched_ctx_get_available_cpuids(unsigned sched_ctx_id, int **cpuids, int *ncpuids)
  2155. {
  2156. int current_worker_id = starpu_worker_get_id();
  2157. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2158. struct starpu_worker_collection *workers = sched_ctx->workers;
  2159. _STARPU_MALLOC((*cpuids), workers->nworkers*sizeof(int));
  2160. int w = 0;
  2161. struct starpu_sched_ctx_iterator it;
  2162. workers->init_iterator(workers, &it);
  2163. while(workers->has_next(workers, &it))
  2164. {
  2165. int workerid = workers->get_next(workers, &it);
  2166. int master = sched_ctx->main_master;
  2167. if(master == current_worker_id || workerid == current_worker_id || current_worker_id == -1)
  2168. {
  2169. (*cpuids)[w++] = starpu_worker_get_bindid(workerid);
  2170. }
  2171. }
  2172. *ncpuids = w;
  2173. return;
  2174. }
  2175. static void _starpu_sched_ctx_put_new_master(unsigned sched_ctx_id)
  2176. {
  2177. int *workerids;
  2178. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2179. unsigned nworkers = starpu_sched_ctx_get_workers_list_raw(sched_ctx_id, &workerids);
  2180. unsigned i;
  2181. for (i=0; i<nworkers; i++)
  2182. {
  2183. if (starpu_worker_get_type(workerids[i]) == STARPU_CPU_WORKER)
  2184. {
  2185. sched_ctx->main_master = workerids[i];
  2186. break;
  2187. }
  2188. }
  2189. }
  2190. struct starpu_perfmodel_arch * _starpu_sched_ctx_get_perf_archtype(unsigned sched_ctx_id)
  2191. {
  2192. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2193. return &sched_ctx->perf_arch;
  2194. }
  2195. int starpu_sched_ctx_get_worker_rank(unsigned sched_ctx_id)
  2196. {
  2197. int idx = 0;
  2198. int curr_workerid = starpu_worker_get_id();
  2199. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2200. if(sched_ctx->sched_policy || !sched_ctx->awake_workers)
  2201. return -1;
  2202. struct starpu_worker_collection *workers = sched_ctx->workers;
  2203. struct starpu_sched_ctx_iterator it;
  2204. workers->init_iterator(workers, &it);
  2205. while(workers->has_next(workers, &it))
  2206. {
  2207. int worker = workers->get_next(workers, &it);
  2208. if(worker == curr_workerid)
  2209. return idx;
  2210. idx++;
  2211. }
  2212. return -1;
  2213. }
  2214. void (*starpu_sched_ctx_get_sched_policy_init(unsigned sched_ctx_id))(unsigned)
  2215. {
  2216. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2217. return sched_ctx->init_sched;
  2218. }
  2219. unsigned starpu_sched_ctx_has_starpu_scheduler(unsigned sched_ctx_id, unsigned *awake_workers)
  2220. {
  2221. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2222. *awake_workers = sched_ctx->awake_workers;
  2223. return sched_ctx->sched_policy != NULL;
  2224. }
  2225. void *starpu_sched_ctx_get_user_data(unsigned sched_ctx_id)
  2226. {
  2227. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  2228. STARPU_ASSERT(sched_ctx != NULL);
  2229. return sched_ctx->user_data;
  2230. }