sched_ctx.c 81 KB

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